Consommer des Messages depuis Apache Kafka¶

Vue d'ensemble¶

Apprenez à construire des applications consumer Kafka qui lisent et traitent des flux d'événements depuis les topics Kafka. Ce guide couvre la configuration des consumers, la gestion des offsets, les consumer groups pour le traitement parallèle, et les meilleures pratiques pour construire des applications de streaming fiables.

!!! info "Ce que vous allez apprendre" - Configurer et créer des consumers Kafka - S'abonner à des topics et interroger pour des messages - Gérer les offsets des consumers pour la tolérance aux pannes - Mettre à l'échelle avec les consumer groups - Implémenter la gestion d'erreurs et la récupération

Prérequis¶

• Docker et Docker Compose installés
• Compréhension de base des topics et partitions Kafka
• Environnement de développement avec les bibliothèques client Kafka

Démarrage Rapide : Exécuter Kafka avec Docker Compose¶

Utilisez cette configuration Docker Compose pour exécuter un cluster Kafka à nœud unique avec KRaft (pas besoin de Zookeeper) :

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

version: "3.8"

services:
  kafka:
    image: confluentinc/cp-kafka:latest # (1)!
    container_name: kafka-kraft
    ports:
      - "9092:9092" # (2)!
    environment:
      # Paramètres KRaft
      KAFKA_NODE_ID: 1 # (3)!
      KAFKA_PROCESS_ROLES: broker,controller # (4)!
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 # (5)!
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 # (6)!
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093 # (7)!

      # ID du cluster (générer avec: kafka-storage random-uuid)
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk # (8)!

      # Stockage
      KAFKA_LOG_DIRS: /var/lib/kafka/data # (9)!

      # Désactiver les métriques Confluent (optionnel pour configuration minimale)
      KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE: "false" # (10)!
    volumes:
      - kafka-data:/var/lib/kafka/data
    healthcheck:
      test: ["CMD-SHELL", "kafka-broker-api-versions --bootstrap-server localhost:9092"]
      interval: 10s
      timeout: 5s
      retries: 5

volumes:
  kafka-data:
    driver: local

1. Image Confluent Platform Kafka (inclut le mode KRaft)
2. Port du broker Kafka exposé à l'hôte
3. ID de nœud unique dans le cluster KRaft
4. Ce nœud agit à la fois comme broker et controller (mode combiné)
5. Listeners : PLAINTEXT pour les clients, CONTROLLER pour la communication interne du cluster
6. Listener annoncé pour les clients externes
7. Votants du quorum pour le consensus KRaft (format : id@host:port)
8. ID de cluster unique - générer avec docker run confluentinc/cp-kafka kafka-storage random-uuid
9. Répertoire de logs pour le stockage des données Kafka
10. Désactiver la collecte de métriques Confluent Support pour une configuration locale minimale

Démarrer Kafka¶

# Démarrer Kafka
docker-compose up -d

# Attendre que Kafka soit prêt
docker-compose ps

# Créer un topic avec quelques données de test
docker exec kafka-kraft kafka-topics \
  --bootstrap-server localhost:9092 \
  --create --topic orders --partitions 3 --replication-factor 1

# Produire des messages de test
docker exec -i kafka-kraft kafka-console-producer \
  --bootstrap-server localhost:9092 \
  --topic orders << EOF
{"order_id": "ORD-001", "amount": 99.99}
{"order_id": "ORD-002", "amount": 149.50}
{"order_id": "ORD-003", "amount": 75.25}
EOF

# Arrêter Kafka
docker-compose down

bash # Consommer les messages depuis le début docker exec kafka-kraft kafka-console-consumer \ --bootstrap-server localhost:9092 \ --topic orders --from-beginning --max-messages 3

Fondamentaux des Consumers¶

Qu'est-ce qu'un Consumer Kafka ?¶

Un consumer est une application client qui lit des données depuis les topics Kafka. Les consumers :

- Récupèrent les messages d'un ou plusieurs topics
- Traitent les événements en temps réel ou quasi-temps réel
- Suivent leur position de lecture avec les offsets
- Peuvent évoluer horizontalement avec les consumer groups

graph LR
    T[Topic: orders<br/>3 Partitions] --> C1[Consumer 1<br/>Lit Partition 0]
    T --> C2[Consumer 2<br/>Lit Partition 1]
    T --> C3[Consumer 3<br/>Lit Partition 2]

    C1 --> P[Logique de<br/>Traitement]
    C2 --> P
    C3 --> P

Étape 1 : Configurer Votre Consumer¶

Configuration Requise¶

Tous les consumers ont besoin d'une configuration de base pour se connecter au cluster Kafka :

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Properties;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");           // (1)!
props.put("group.id", "order-processing-service");          // (2)!
props.put("key.deserializer", StringDeserializer.class);    // (3)!
props.put("value.deserializer", StringDeserializer.class);  // (4)!
props.put("auto.offset.reset", "earliest");                 // (5)!

1
2
3
4
5
6
7
8
9

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    bootstrap_servers='localhost:9092',        # (1)!
    group_id='order-processing-service',       # (2)!
    key_deserializer=lambda k: k.decode('utf-8'),   # (3)!
    value_deserializer=lambda v: v.decode('utf-8'), # (4)!
    auto_offset_reset='earliest'               # (5)!
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'order-processing-service',
  brokers: ['localhost:9092']              // (1)!
});

const consumer = kafka.consumer({
  groupId: 'order-processing-service'      // (2)!
});

// Auto-désérialisation en string par défaut
// Configurer dans la boucle de consommation si nécessaire  // (3)!

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

package main

import (
    "github.com/confluentinc/confluent-kafka-go/kafka"
)

config := &kafka.ConfigMap{
    "bootstrap.servers": "localhost:9092",          // (1)!
    "group.id":          "order-processing-service", // (2)!
    "auto.offset.reset": "earliest",                 // (3)!
}

consumer, err := kafka.NewConsumer(config)
if err != nil {
    panic(err)
}
defer consumer.Close()

Configuration Optionnelle¶

Considérez ces paramètres optionnels pour la production :

| Paramètre | Description | Défaut | Recommandation |
|-----------|-------------|---------|----------------|
| `fetch.min.bytes` | Données minimales par requête fetch | 1 | Augmenter pour le débit |
| `fetch.max.wait.ms` | Temps d'attente max si min bytes non atteint | 500ms | Équilibrer latence vs débit |
| `max.poll.records` | Records max par poll | 500 | Ajuster selon temps de traitement |
| `enable.auto.commit` | Auto-commit des offsets | `true` | `false` pour contrôle manuel |
| `session.timeout.ms` | Timeout heartbeat du consumer | 10s | Augmenter pour traitement lent |

Étape 2 : S'abonner aux Topics¶

Abonnement à un Topic Unique¶

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// S'abonner à un topic unique
consumer.subscribe(Collections.singletonList("orders"));  // (1)!

# S'abonner à un topic unique
consumer.subscribe(['orders'])  // (1)!

await consumer.connect();  // (1)!
await consumer.subscribe({
  topic: 'orders',         // (2)!
  fromBeginning: true      // (3)!
});

err := consumer.SubscribeTopics([]string{"orders"}, nil)  // (1)!
if err != nil {
    panic(err)
}

Topics Multiples¶

consumer.subscribe(Arrays.asList("orders", "payments", "shipments"));

consumer.subscribe(['orders', 'payments', 'shipments'])

await consumer.subscribe({ topics: ['orders', 'payments', 'shipments'] });

consumer.SubscribeTopics([]string{"orders", "payments", "shipments"}, nil)

Abonnement Basé sur Pattern¶

S'abonner aux topics correspondant à un pattern regex - utile pour la création dynamique de topics mais nécessite une gestion attentive.

import java.util.regex.Pattern;

consumer.subscribe(Pattern.compile("order-.*"));  // (1)!

import re

consumer.subscribe(pattern=re.compile(r'order-.*'))  // (1)!

Étape 3 : Interroger pour des Messages¶

La Boucle de Poll¶

Les applications de streaming s'exécutent indéfiniment, interrogeant continuellement pour de nouveaux messages. C'est le comportement normal pour les systèmes événementiels.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

try {
    while (true) {  // (1)!
        ConsumerRecords<String, String> records =
            consumer.poll(Duration.ofMillis(100));  // (2)!

        for (ConsumerRecord<String, String> record : records) {  // (3)!
            System.out.printf(
                "Topic=%s Partition=%d Offset=%d Key=%s Value=%s%n",
                record.topic(),      // (4)!
                record.partition(),  // (5)!
                record.offset(),     // (6)!
                record.key(),        // (7)!
                record.value()       // (8)!
            );

            // Traitez votre message ici
            processOrder(record.value());
        }
    }
} finally {
    consumer.close();  // (9)!
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

try:
    for message in consumer:  // (1)!
        print(f"Topic={message.topic} "
              f"Partition={message.partition} "
              f"Offset={message.offset} "
              f"Key={message.key} "
              f"Value={message.value}")

        # Traitez votre message ici
        process_order(message.value)  // (2)!

finally:
    consumer.close()  // (3)!

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

await consumer.run({  // (1)!
  eachMessage: async ({ topic, partition, message }) => {  // (2)!
    console.log({
      topic,
      partition,
      offset: message.offset,
      key: message.key?.toString(),      // (3)!
      value: message.value.toString(),   // (4)!
    });

    // Traitez votre message ici
    await processOrder(message.value.toString());  // (5)!
  },
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

for {  // (1)!
    msg, err := consumer.ReadMessage(-1)  // (2)!
    if err != nil {
        fmt.Printf("Consumer error: %v\n", err)
        continue
    }

    fmt.Printf("Topic=%s Partition=%d Offset=%d Key=%s Value=%s\n",
        *msg.TopicPartition.Topic,
        msg.TopicPartition.Partition,
        msg.TopicPartition.Offset,
        string(msg.Key),
        string(msg.Value))

    // Traitez votre message ici
    processOrder(string(msg.Value))  // (3)!
}

Étape 4 : Gérer les Offsets des Consumers¶

Comprendre les Offsets¶

Kafka suit l'offset (position) de chaque message qu'un consumer a traité. Cela permet :

- **Tolérance aux pannes** : Reprendre depuis la dernière position après un crash
- **Sémantique exactly-once** : Éviter de retraiter les messages
- **Rejouabilité** : Revenir à un offset antérieur si nécessaire

sequenceDiagram
    participant C as Consumer
    participant K as Broker Kafka
    participant T as topic __consumer_offsets

    C->>K: Poll messages (offset 100-109)
    K->>C: Retourne 10 messages
    C->>C: Traite les messages
    C->>K: Commit offset 110
    K->>T: Stocke offset 110 pour le groupe

    Note over C: Consumer crash

    C->>K: Consumer redémarre, fetch offset
    T->>K: Retourne offset 110
    K->>C: Reprend depuis offset 110

Auto vs Commit Manuel¶

props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000");

props.put("enable.auto.commit", "false");
consumer.commitSync();  // ou commitAsync()

Exemples de Commit Manuel¶

props.put("enable.auto.commit", "false");  // (1)!

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

    for (ConsumerRecord<String, String> record : records) {
        try {
            processOrder(record.value());

            // Commit après traitement réussi
            consumer.commitSync();  // (2)!
        } catch (Exception e) {
            // Gérer l'erreur - ne pas commiter en cas d'échec
            log.error("Processing failed", e);  // (3)!
        }
    }
}

consumer = KafkaConsumer(
    'orders',
    enable_auto_commit=False  // (1)!
)

for message in consumer:
    try:
        process_order(message.value)
        consumer.commit()  // (2)!
    except Exception as e:
        print(f"Error processing: {e}")  // (3)!
        # Ne pas commiter - réessayera au redémarrage

await consumer.run({
  autoCommit: false,  // (1)!
  eachMessage: async ({ message, heartbeat, pause }) => {
    try {
      await processOrder(message.value.toString());
      await consumer.commitOffsets([{  // (2)!
        topic: 'orders',
        partition: message.partition,
        offset: (parseInt(message.offset) + 1).toString()
      }]);
    } catch (error) {
      console.error('Processing failed:', error);
      pause();  // (3)!
      setTimeout(() => consumer.resume([{ topic: 'orders' }]), 5000);
    }
  }
});

config := &kafka.ConfigMap{
    "enable.auto.commit": false,  // (1)!
}

for {
    msg, err := consumer.ReadMessage(-1)
    if err != nil {
        continue
    }

    if err := processOrder(string(msg.Value)); err == nil {
        _, err := consumer.CommitMessage(msg)  // (2)!
        if err != nil {
            log.Printf("Commit failed: %v\n", err)
        }
    } else {
        log.Printf("Processing failed: %v\n", err)  // (3)!
    }
}

Étape 5 : Mettre à l'Échelle avec les Consumer Groups¶

Concepts des Consumer Groups¶

Les consumer groups permettent le traitement parallèle en distribuant les partitions entre plusieurs instances de consumers.

**Règles Clés :**

- Les consumers dans le même groupe partagent la charge des partitions
- Chaque partition est assignée à **un seul** consumer du groupe
- Plusieurs groupes peuvent lire le même topic indépendamment

graph TB
    subgraph "Topic: orders (3 partitions)"
    P0[Partition 0]
    P1[Partition 1]
    P2[Partition 2]
    end

    subgraph "Consumer Group: order-processing"
    C1[Consumer 1]
    C2[Consumer 2]
    C3[Consumer 3]
    end

    P0 --> C1
    P1 --> C2
    P2 --> C3

Rééquilibrage¶

Quand des consumers rejoignent ou quittent un groupe, Kafka rééquilibre automatiquement les assignations de partitions.

**Déclencheurs :**

- Un nouveau consumer rejoint le groupe
- Un consumer existant échoue/s'arrête
- Le nombre de partitions du topic change

**Pendant le rééquilibrage :**

- Courte pause de traitement (habituellement < 1 seconde)
- Partitions réassignées aux consumers disponibles

graph TB
    subgraph "Avant Rééquilibrage"
    P0A[Partition 0] --> C1A[Consumer 1]
    P1A[Partition 1] --> C2A[Consumer 2 ❌ ÉCHOUE]
    P2A[Partition 2] --> C3A[Consumer 3]
    end

    subgraph "Après Rééquilibrage"
    P0B[Partition 0] --> C1B[Consumer 1]
    P1B[Partition 1] --> C1B
    P2B[Partition 2] --> C3B[Consumer 3]
    end

Directives de Mise à l'Échelle¶

Exemple :

• Topic avec 6 partitions → Déployer 6 consumers pour débit maximum
• Topic avec 3 partitions → Ne pas déployer plus de 3 consumers

Gestion d'Erreurs et Meilleures Pratiques¶

Gérer les Échecs de Traitement¶

1. Erreurs Réseau Transitoires ```java int maxRetries = 3; int retryCount = 0;

while (retryCount < maxRetries) {
    try {
        processOrder(record.value());
        consumer.commitSync();
        break;  // Succès
    } catch (TransientException e) {
        retryCount++;
        Thread.sleep(1000 * retryCount);  // Backoff exponentiel
    }
}
```

**2. Poison Pills (Messages Corrompus)**
```java
try {
    processOrder(record.value());
} catch (DeserializationException e) {
    // Envoyer vers dead letter queue
    producer.send(new ProducerRecord<>("orders-dlq", record.value()));
    consumer.commitSync();  // Ignorer le mauvais message
}
```

**3. Service Downstream Indisponible**
```java
try {
    sendToDatabase(record.value());
} catch (DatabaseException e) {
    consumer.pause(Collections.singleton(
        new TopicPartition(record.topic(), record.partition())
    ));
    // Réessayer après délai
    scheduleRetry();
}
```

Liste de Contrôle des Meilleures Pratiques¶

!!! success "Consumers Prêts pour la Production" - [x] Utiliser les commits d'offset manuels pour données critiques - [x] Implémenter un traitement idempotent (gérer les duplicatas) - [x] Définir un session.timeout.ms approprié pour votre charge de travail - [x] Surveiller les métriques de lag des consumers - [x] Gérer les rééquilibrages gracieusement - [x] Utiliser des dead letter queues pour poison pills - [x] Implémenter des hooks d'arrêt appropriés - [x] Logger les offsets et métadonnées de traitement

Arrêt Gracieux¶

Exemple Complet : Service de Traitement de Commandes¶

Voici un exemple de consumer prêt pour la production :

public OrderConsumerService(Properties config) {
    this.consumer = new KafkaConsumer<>(config);
}

public void start() {
    Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));

    consumer.subscribe(Collections.singletonList("orders"));
    System.out.println("Consumer started, waiting for messages...");

    try {
        while (running) {
            ConsumerRecords<String, String> records =
                consumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, String> record : records) {
                processRecord(record);
            }

            // Commit après le batch
            consumer.commitSync();
        }
    } catch (WakeupException e) {
        // Attendu pendant l'arrêt
    } catch (Exception e) {
        System.err.println("Consumer error: " + e.getMessage());
    } finally {
        consumer.close();
        System.out.println("Consumer closed");
    }
}

private void processRecord(ConsumerRecord<String, String> record) {
    try {
        System.out.printf("Processing order: %s = %s%n",
            record.key(), record.value());

        // Votre logique métier ici
        // ex : parser JSON, valider, stocker en base de données

    } catch (Exception e) {
        System.err.println("Failed to process record: " + e.getMessage());
        // Envoyer vers DLQ ou implémenter logique de retry
    }
}

private void shutdown() {
    System.out.println("Shutdown signal received");
    running = false;
    consumer.wakeup();
}

public static void main(String[] args) {
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("group.id", "order-processing-service");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("enable.auto.commit", "false");
    props.put("auto.offset.reset", "earliest");

    OrderConsumerService service = new OrderConsumerService(props);
    service.start();
}

def start(self):
    signal.signal(signal.SIGINT, self._shutdown)
    signal.signal(signal.SIGTERM, self._shutdown)

    print("Consumer started, waiting for messages...")

    try:
        for message in self.consumer:
            if not self.running:
                break
            self._process_message(message)
            self.consumer.commit()
    except Exception as e:
        print(f"Consumer error: {e}")
    finally:
        self.consumer.close()
        print("Consumer closed")

def _process_message(self, message):
    try:
        order = message.value
        print(f"Processing order: {message.key} = {order}")

        # Votre logique métier ici
        # ex : valider commande, mettre à jour inventaire, envoyer notification

    except Exception as e:
        print(f"Failed to process message: {e}")
        # Envoyer vers DLQ ou implémenter logique de retry

def _shutdown(self, signum, frame):
    print("Shutdown signal received")
    self.running = False

service = OrderConsumerService(config)
service.start()

Prochaines Étapes¶

!!! tip "Continuez Votre Apprentissage" - Utiliser Schema Registry - Gérer les schémas de données pour la sécurité des types - Connecter des Systèmes Externes - Intégrer Kafka avec des bases de données - Traiter des Streams - Transformer des données en temps réel - Retour au Tutoriel - Réviser les concepts fondamentaux

Ressources Supplémentaires¶

Documentation Officielle¶

• Documentation API Apache Kafka Consumer
• Référence Configuration Kafka Consumer
• Protocole Consumer Group - Détails Internes
• Guide Configuration Consumer Confluent

Tutoriels & Cours¶

• Cours Kafka Consumer 101 - Cours gratuit Confluent
• Exercice Interactif Kafka Consumer
• Meilleures Pratiques Consumer

Sujets Avancés¶

• Optimisation Performance Consumer
• Rééquilibrage Consumer Group - Analyse Approfondie
• Stratégies de Gestion des Offsets

Ce guide est basé sur le contenu du cours Apache Kafka 101 de Confluent et la documentation officielle Apache Kafka.