kafka ajout DLQ et retry

Author: ErwanLT

messaging-tutorial/kafka-tutorial/kafka-receiver/README.md

@@ -27,4 +27,35 @@ Assurez-vous que Kafka est démarré (via `docker-compose up -d` dans le dossier
 2.  Exécutez `mvn spring-boot:run`.
 3.  L'application va démarrer et commencer à écouter les messages. Envoyez des messages via l'application `kafka-sender` pour les voir apparaître ici.
     *   Envoyez un message normal (ex: `{"message": "Hello Kafka!"}`) : Il sera traité normalement.
-    *   Envoyez un message contenant "erreur" (ex: `{"message": "Ceci est une erreur"}`) : Il sera loggé comme une erreur par le consommateur principal et **vous verrez un log indiquant qu'il a été reçu par le listener de la DLQ**.
+    *   Envoyez un message contenant "erreur" (ex: `{"message": "Ceci est une erreur"}`) : # Application Consommateur Kafka (kafka-receiver)
+
+Cette application Spring Boot agit comme un **consommateur** de messages pour Apache Kafka.
+
+## Description
+
+Son unique rôle est d'écouter en continu un topic Kafka spécifique. Lorsqu'un message est publié dans ce topic (par l'application `kafka-sender`), cette application le consomme et l'affiche dans la console.
+
+Elle démontre la manière la plus simple de recevoir des messages avec Spring Kafka.
+
+## Concepts Clés
+
+-   **Spring Kafka `@KafkaListener`** : L'annotation qui transforme une simple méthode en un consommateur de messages Kafka puissant et managé par Spring.
+-   **Consommateur Kafka** : Le rôle de cette application dans l'écosystème Kafka.
+-   **Dead Letter Queue (DLQ)** : Un mécanisme pour gérer les messages en erreur, les redirigeant vers un topic dédié (`my-topic.DLT`).
+
+## Fichiers Principaux
+
+-   `KafkaConsumerService.java` : Contient la méthode annotée avec `@KafkaListener` qui est automatiquement invoquée pour chaque nouveau message dans le topic, ainsi que le listener pour la DLQ.
+-   `KafkaConsumerConfig.java` : Configure le `ConcurrentKafkaListenerContainerFactory` et le `DefaultErrorHandler` pour la gestion des DLQ.
+
+## Comment l'exécuter
+
+Assurez-vous que Kafka est démarré (via `docker-compose up -d` dans le dossier parent).
+
+1.  Naviguez dans ce dossier.
+2.  Exécutez `mvn spring-boot:run`.
+3.  L'application va démarrer et commencer à écouter les messages. Envoyez des messages via l'application `kafka-sender` pour les voir apparaître ici.
+    *   Envoyez un message normal (ex: `{"message": "Hello Kafka!"}`) : Il sera traité normalement.
+    *   Envoyez un message contenant "erreur" (ex: `{"message": "Ceci est une erreur"}`) : Il sera loggé comme une erreur par le consommateur principal, envoyé à la DLQ, et **vous verrez un log indiquant qu'il a été reçu par le listener de la DLQ pour une tentative de re-traitement**.
+    *   Envoyez un message contenant "re-erreur" (ex: `{"message": "Ceci est une re-erreur"}`) : Il sera d'abord envoyé à la DLQ, puis le listener de la DLQ tentera de le re-traiter, échouera à nouveau, et **vous verrez un log indiquant l'échec du re-traitement depuis la DLQ**.
+

messaging-tutorial/kafka-tutorial/kafka-receiver/src/main/java/fr/eletutour/receiver/service/KafkaConsumerService.java

@@ -18,9 +18,11 @@ public class KafkaConsumerService {
     )
     public void listen(String message, Acknowledgment acknowledgment) {
         LOG.info("Message reçu : {}", message);
-
+        
+        // Simuler une erreur pour les messages contenant "erreur"
         if (message.contains("erreur")) {
             LOG.error("Simulating processing error for message: {}", message);
+            // L'exception sera interceptée par le DefaultErrorHandler
             throw new RuntimeException("Erreur de traitement simulée pour le message: " + message);
         }
 
@@ -29,13 +31,30 @@ public void listen(String message, Acknowledgment acknowledgment) {
         acknowledgment.acknowledge(); // Commit manuel de l'offset
     }
 
+    // Listener pour la Dead Letter Queue (DLQ)
     @KafkaListener(
             topics = "${app.kafka.dlt-topic}",
             groupId = "${spring.kafka.consumer.group-id}",
             containerFactory = "kafkaListenerContainerFactory"
     )
     public void listenDlt(String message, Acknowledgment acknowledgment) {
         LOG.warn("Message reçu de la DLQ : {}", message);
-        acknowledgment.acknowledge();
+        
+        try {
+            // Simuler une tentative de re-traitement
+            LOG.info("Tentative de re-traitement du message de la DLQ : {}", message);
+            
+            // Pour la démonstration, si le message contient "re-erreur", il échoue à nouveau
+            if (message.contains("re-erreur")) {
+                throw new RuntimeException("Échec du re-traitement simulé pour le message: " + message);
+            }
+
+            // Si le re-traitement réussit
+            LOG.info("Message de la DLQ re-traité avec succès : {}", message);
+            acknowledgment.acknowledge(); // Acknowledge le message de la DLQ
+        } catch (Exception e) {
+            LOG.error("Échec du re-traitement du message de la DLQ : {}", message, e);
+            acknowledgment.acknowledge();
+        }
     }
 }