⚡ Apache Kafka – Architecture, Opérations & Bonnes Pratiques

[Producer] --batch--> [Broker/Leader] --replicate--> [Followers]
[Segments .log] <= retention/compactage
[Consumer Group] --fetch--> [Partition 0..n] --commit--> __consumer_offsets

Le débit provient du séquencement disque + envoi séquentiel réseau, et du batching.

Règles de pouce (rough sizing)

• Partitions : ~ (10 × nb. consommateurs parallèles visés).
• Stockage : débit_journalier × rétention_jours × réplication (×1.2 marge).
• RF : 3 pour prod. min.insync.replicas = 2 + acks=all.

min.insync.replicas=2
acks=all   # requiert ≥2 ISR pour l’ack
unclean.leader.election.enable=false
        

Surveiller UnderReplicatedPartitions; déclencher un reassignment si nécessaire.

process.roles=broker,controller
controller.listener.names=CONTROLLER
controller.quorum.voters=1@node1:9093,2@node2:9093,3@node3:9093
        

• Pas de ZooKeeper : bootstrapping via cluster.id.
• 3 voters min. Production : 3 ou 5 selon SLA.

kafka-reassign-partitions.sh --generate --topics-to-move-json-file topics.json --broker-list "1,2,3" --bootstrap-server :9092
kafka-reassign-partitions.sh --execute --reassignment-json-file map.json --bootstrap-server :9092
        

Utiliser --verify et surveiller le throttle pour limiter l’impact.

+ Controller Quorum (KRaft)
| voters: node1,node2,node3
+-----v----+   replicate   +-----------+
| Broker 1 |<------------->| Broker 2  |
+----------+               +-----------+
        

docker‑compose.yml

version: '3.8'
services:
  kafka:
    image: bitnami/kafka:3.7
    environment:
      - KAFKA_ENABLE_KRAFT=yes
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      - KAFKA_CFG_NODE_ID=1
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
      - ALLOW_PLAINTEXT_LISTENER=yes
    ports: ["9092:9092","9093:9093"]
    volumes: ["kdata:/bitnami/kafka"]
volumes: { kdata: {} }
        

# KRaft cluster (3 voters)
CLUSTER_ID=$(bin/kafka-storage.sh random-uuid)
# init + format + start (node1..node3)
        

• Configurer advertised.listeners pour clients distants.
• RAID1/10 conseillé, XFS/ext4.

• kcat (ancien kafkacat) : production/consommation rapide.
• jmx_exporter + Prometheus + Grafana.
• Cruise Control : rebalancing automatisé.

Properties p=new Properties();
p.put("bootstrap.servers","broker:9092");
p.put("acks","all");
p.put("enable.idempotence",true);
p.put("linger.ms",10);
Producer prod=new KafkaProducer<>(p,new StringSerializer(), new ByteArraySerializer());
prod.send(new ProducerRecord<>("orders","k1", payload));
        

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=['broker:9092'],
                         acks='all', linger_ms=10, value_serializer=lambda v: v)
producer.send('orders', key=b'k1', value=b'{...}')
producer.flush()
        

# Conseils débit
- Favoriser des messages 1–10 KB, mais batcher jusqu'à 128–512 KB.
- Multi‑threads producteurs (1 conn / plusieurs threads ok).
- Eviter l'ACK local (acks=1) en prod; préférer acks=all.
        

partition.assignment.strategy=cooperative-sticky
# réduit les pauses; nécessite clients récents
        

kafka-consumer-groups.sh --bootstrap-server :9092 --describe --group g1
# Exporter le lag vers Prometheus (burrow, kafka-lag-exporter)

partitioner: key -> hash(key) % partitions
# pour l’ordre par clé, réutiliser la même clé (ex: userId)

enable.idempotence=true
transactional.id=service-A
# côté consumer: read-process-produce-commit transactionnel

State local (RocksDB) + changelog Kafka. Sauvegarde compacte, rééquilibrage en cas de rebalance.

orders.join(payments, joiner, JoinWindows.ofSeconds(30))

bootstrap.servers=broker:9092
config.storage.topic=connect-configs
status.storage.topic=connect-status
offset.storage.topic=connect-offsets
        

errors.tolerance=all
errors.deadletterqueue.topic.name=_connect_dlq
transforms=unwrap
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
        

compatibility.level=BACKWARD
# Process d’évolution: add fields with default, éviter les breaks

sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
listener.name.sasl_ssl.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required ...;

kafka-acls.sh --bootstrap-server :9092 --add --allow-principal User:alice --operation READ --topic orders
        

socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
num.network.threads=8
        

num.io.threads=16
queued.max.requests=1000
replica.fetch.max.bytes=1048576
        

# Symptômes: pauses longues, throughput en dents de scie
- Activer cooperative-sticky
- Réduire max.poll.interval.ms si trop grand
- Augmenter partitions (scaling horizontal)
        

du -sh /kafka-logs/*
# Purges manuelles contrôlées, vérifier retention/segment