Master 2 GLIA Notes

TP6 - Kafka

9 min de lecture

Étape 1 : Installation et configuration de Kafka

1. Téléchargement de Kafka

Téléchargement de la dernière version stable depuis le site officiel Apache Kafka. Extraction de l’archive dans le répertoire d’installation.

2. Génération d’un cluster ID

Pour Kafka en mode KRaft (sans ZooKeeper), génération d’un identifiant unique de cluster :

sudo /usr/share/kafka/bin/kafka-storage.sh random-uuid

Résultat obtenu :

Jp8Wc3R6QH2sKZyA9LxM5D

Cet identifiant sera utilisé pour formater les répertoires de stockage.

3. Formatage des répertoires de stockage

Initialisation du stockage Kafka avec l’ID de cluster généré :

sudo /usr/share/kafka/bin/kafka-storage.sh format -t Jp8Wc3R6QH2sKZyA9LxM5D -c /etc/kafka/server.properties --standalone

Résultat :

Formatting dynamic metadata voter directory /var/lib/kafka with metadata.version 4.1-IV1.

Attribution des permissions au dossier Kafka :

sudo chown -R kafka:kafka /var/lib/kafka

4. Démarrage de Kafka

Lancement du service Kafka via systemd :

sudo systemctl start kafka

Vérification du statut du service :

sudo systemctl status kafka

Sortie obtenue :

kafka.service - Kafka server
     Loaded: loaded (/usr/lib/systemd/system/kafka.service; disabled; preset: disabled)
     Active: active (running) since Sat 2025-12-20 23:15:29 CET; 4s ago
   Main PID: 229314 (java)
      Tasks: 111 (limit: 18810)
     Memory: 410.7M (peak: 411.8M)
        CPU: 6.068s

Le service Kafka est actif et fonctionnel. Le processus Java consomme environ 410 MB de mémoire.

5. Création du topic commande-creee

Création d’un topic Kafka avec une partition et un facteur de réplication de 1 :

/usr/share/kafka/bin/kafka-topics.sh --create --topic commande-creee --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

Résultat :

Created topic commande-creee.

Vérification de la création du topic :

/usr/share/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

Liste des topics :

__consumer_offsets
commande-creee

Le topic commande-creee est bien créé. Le topic __consumer_offsets est créé automatiquement par Kafka pour gérer les offsets des consommateurs.

Étape 2 : Développement du service Produit

1. Création du projet Maven

Création d’un projet Spring Boot nommé produitService avec les dépendances suivantes dans pom.xml :

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
        <scope>runtime</scope>
    </dependency>
</dependencies>

2. Entité Produit

Création de l’entité Produit.java dans le package com.produit.model :

package com.produit.model;
 
import javax.persistence.*;
 
@Entity
@Table(name = "produits")
public class Produit {
    
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(nullable = false)
    private String nom;
    
    @Column(nullable = false)
    private double prix;
    
    public Produit() {}
    
    public Produit(String nom, double prix) {
        this.nom = nom;
        this.prix = prix;
    }
    
    // Getters et setters
    public Long getId() {
        return id;
    }
    
    public void setId(Long id) {
        this.id = id;
    }
    
    public String getNom() {
        return nom;
    }
    
    public void setNom(String nom) {
        this.nom = nom;
    }
    
    public double getPrix() {
        return prix;
    }
    
    public void setPrix(double prix) {
        this.prix = prix;
    }
}

3. Repository Produit

Création de ProduitRepository.java dans le package com.produit.repository :

package com.produit.repository;
 
import com.produit.model.Produit;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;
 
@Repository
public interface ProduitRepository extends JpaRepository<Produit, Long> {
}

L’interface étend JpaRepository qui fournit toutes les opérations CRUD de base.

4. Contrôleur REST Produit

Création de ProduitController.java dans le package com.produit.controller :

package com.produit.controller;
 
import com.produit.model.Produit;
import com.produit.repository.ProduitRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.util.List;
 
@RestController
@RequestMapping("/produits")
public class ProduitController {
    
    @Autowired
    private ProduitRepository produitRepository;
    
    @GetMapping
    public List<Produit> getAllProduits() {
        return produitRepository.findAll();
    }
    
    @PostMapping
    public ResponseEntity<Produit> createProduit(@RequestBody Produit produit) {
        Produit savedProduit = produitRepository.save(produit);
        return ResponseEntity.ok(savedProduit);
    }
}

Endpoints exposés :

  • GET /produits : Récupère la liste de tous les produits
  • POST /produits : Crée un nouveau produit

5. Configuration

Fichier application.properties :

server.port=8082
spring.application.name=produit-service
 
# H2 Database
spring.datasource.url=jdbc:h2:mem:produitDB
spring.datasource.driverClassName=org.h2.Driver
spring.datasource.username=sa
spring.datasource.password=password
 
# JPA
spring.jpa.hibernate.ddl-auto=create-drop
spring.jpa.show-sql=true
 
# H2 Console
spring.h2.console.enabled=true
spring.h2.console.path=/h2-console

Le service Produit est configuré sur le port 8082 avec une base H2 en mémoire. La console H2 est accessible pour vérifier les données.

Classe principale ProduitServiceApplication.java :

package com.produit;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
public class ProduitServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(ProduitServiceApplication.class, args);
    }
}

Étape 3 : Développement du service Commande

1. Création du projet Maven

Création du projet commandeService avec les dépendances Spring Boot Web, JPA, Kafka et H2 dans pom.xml :

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
        <scope>runtime</scope>
    </dependency>
</dependencies>

2. Entité Commande

Création de l’entité Commande.java dans le package com.commande.model :

package com.commande.model;
 
import javax.persistence.*;
 
@Entity
@Table(name = "commandes")
public class Commande {
    
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(nullable = false)
    private Long produitId;
    
    @Column(nullable = false)
    private int quantite;
    
    public Commande() {}
    
    public Commande(Long produitId, int quantite) {
        this.produitId = produitId;
        this.quantite = quantite;
    }
    
    // Getters et setters
    public Long getId() {
        return id;
    }
    
    public void setId(Long id) {
        this.id = id;
    }
    
    public Long getProduitId() {
        return produitId;
    }
    
    public void setProduitId(Long produitId) {
        this.produitId = produitId;
    }
    
    public int getQuantite() {
        return quantite;
    }
    
    public void setQuantite(int quantite) {
        this.quantite = quantite;
    }
}

3. Repository Commande

Création de CommandeRepository.java dans le package com.commande.repository :

package com.commande.repository;
 
import com.commande.model.Commande;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;
 
@Repository
public interface CommandeRepository extends JpaRepository<Commande, Long> {
}

4. Service Commande avec producteur Kafka

Création de CommandeService.java dans le package com.commande.service :

package com.commande.service;
 
import com.commande.model.Commande;
import com.commande.repository.CommandeRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
 
@Service
public class CommandeService {
    
    @Autowired
    private CommandeRepository commandeRepository;
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    private static final String TOPIC = "commande-creee";
    
    public Commande createCommande(Long produitId, int quantite) {
        Commande commande = new Commande(produitId, quantite);
        Commande savedCommande = commandeRepository.save(commande);
        
        // Publication d'un événement Kafka après la création
        String message = "Commande ID: " + savedCommande.getId() + 
                        ", Produit ID: " + savedCommande.getProduitId() + 
                        ", Quantité: " + savedCommande.getQuantite();
        
        kafkaTemplate.send(TOPIC, message);
        
        return savedCommande;
    }
}

Rôle du service : Après avoir persisté la commande en base H2, le service publie un message sur le topic Kafka commande-creee. Le KafkaTemplate est injecté automatiquement par Spring Boot.

5. Contrôleur REST Commande

Création de CommandeController.java dans le package com.commande.controller :

package com.commande.controller;
 
import com.commande.model.Commande;
import com.commande.service.CommandeService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.util.Map;
 
@RestController
@RequestMapping("/commandes")
public class CommandeController {
    
    @Autowired
    private CommandeService commandeService;
    
    @PostMapping
    public ResponseEntity<Commande> createCommande(@RequestBody Map<String, Object> commandeData) {
        Long produitId = Long.valueOf(commandeData.get("produitId").toString());
        int quantite = Integer.parseInt(commandeData.get("quantite").toString());
        
        Commande commande = commandeService.createCommande(produitId, quantite);
        return ResponseEntity.ok(commande);
    }
}

Endpoint exposé : POST /commandes pour créer une commande avec produitId et quantite.

6. Configuration

Fichier application.properties :

server.port=8081
spring.application.name=commande-service
 
# H2 Database
spring.datasource.url=jdbc:h2:mem:commandeDB
spring.datasource.driverClassName=org.h2.Driver
spring.datasource.username=sa
spring.datasource.password=password
 
# JPA
spring.jpa.hibernate.ddl-auto=create-drop
spring.jpa.show-sql=true
 
# Kafka
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

Configuration Kafka : Le service se connecte au broker Kafka sur localhost:9092 et utilise des sérialiseurs de type String pour les clés et valeurs.

Classe principale CommandeServiceApplication.java :

package com.commande;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
public class CommandeServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(CommandeServiceApplication.class, args);
    }
}

Étape 4 : Développement du service Notification

1. Création du projet Maven

Création du projet notificationService avec les dépendances Spring Boot Web et Spring Kafka dans pom.xml :

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
</dependencies>

Ce service ne nécessite pas de base de données car il consomme uniquement des messages Kafka.

2. Listener Kafka

Création de NotificationListener.java dans le package com.notification.listener :

package com.notification.listener;
 
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
 
@Component
public class NotificationListener {
    
    @KafkaListener(topics = "commande-creee", groupId = "notification-service-group")
    public void handleCommandeCreee(String message) {
        System.out.println("========================================");
        System.out.println("Notification reçue !");
        System.out.println("Message: " + message);
        System.out.println("========================================");
    }
}

Annotation @KafkaListener : Configure le listener pour écouter le topic commande-creee avec le groupe de consommateurs notification-service-group.

Comportement : Dès qu’un message est publié sur le topic commande-creee, la méthode handleCommandeCreee est invoquée automatiquement et affiche le message dans les logs.

3. Configuration

Fichier application.properties :

server.port=8083
spring.application.name=notification-service
 
# Kafka
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.group-id=notification-service-group

Configuration Kafka Consumer : Le service se connecte au broker Kafka et utilise des désérialiseurs String. Le groupe de consommateurs permet de gérer les offsets et d’assurer qu’un message n’est traité qu’une seule fois.

Classe principale NotificationServiceApplication.java :

package com.notification;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
public class NotificationServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(NotificationServiceApplication.class, args);
    }
}

Étape 5 : Tests et résultats

1. Démarrage des services

Lancement des trois microservices dans l’ordre :

# Terminal 1 - Produit Service
cd produitService
mvn spring-boot:run
 
# Terminal 2 - Commande Service
cd commandeService
mvn spring-boot:run
 
# Terminal 3 - Notification Service
cd notificationService
mvn spring-boot:run

Les trois services démarrent correctement sur leurs ports respectifs.

2. Test avec Postman

Test 1 : Création d’un produit

Requête POST vers http://localhost:8082/produits :

{
    "nom": "Laptop Dell XPS 15",
    "prix": 1299.99
}

Réponse obtenue :

{
    "id": 1,
    "nom": "Laptop Dell XPS 15",
    "prix": 1299.99
}

Le produit est créé avec succès et stocké dans la base H2 du service Produit.

Test 2 : Récupération des produits

Requête GET vers http://localhost:8082/produits :

Réponse :

[
    {
        "id": 1,
        "nom": "Laptop Dell XPS 15",
        "prix": 1299.99
    }
]

La liste des produits est récupérée correctement.

Test 3 : Création d’une commande

Requête POST vers http://localhost:8081/commandes :

{
    "produitId": 1,
    "quantite": 2
}

Réponse obtenue :

{
    "id": 1,
    "produitId": 1,
    "quantite": 2
}

La commande est créée et persistée dans la base H2 du service Commande.

3. Vérification des logs du service Notification

Immédiatement après la création de la commande, les logs du notificationService affichent :

========================================
Notification reçue !
Message: Commande ID: 1, Produit ID: 1, Quantité: 2
========================================

La communication asynchrone via Kafka fonctionne correctement. Le message publié par commandeService est consommé instantanément par notificationService.

Test 4 : Création d’une seconde commande

Requête POST vers http://localhost:8081/commandes :

{
    "produitId": 1,
    "quantite": 5
}

Réponse :

{
    "id": 2,
    "produitId": 1,
    "quantite": 5
}

Logs du service Notification :

========================================
Notification reçue !
Message: Commande ID: 2, Produit ID: 1, Quantité: 5
========================================

Le système gère correctement plusieurs commandes successives. Chaque commande déclenche bien une notification.

4. Vérification dans la console H2

Accès à la console H2 du service Commande via http://localhost:8081/h2-console :

Connexion avec les paramètres configurés (JDBC URL: jdbc:h2:mem:commandeDB, username: sa, password: password).

Requête SQL :

SELECT * FROM commandes;

Résultat :

ID  | PRODUIT_ID | QUANTITE
----+------------+---------
1   | 1          | 2
2   | 1          | 5

Les commandes sont bien persistées dans la base H2. La synchronisation entre la persistence et la publication Kafka fonctionne correctement.

Vue Graphique

Liens retour