Databevrijding: De native cluster mirroring van Apache Kafka

Apache Kafka is uitstekend in het verplaatsen van gegevens binnen een cluster. Leaders repliceren naar followers, consumers halen gegevens op van elke replica en het hele mechanisme draait met minimale operationele overhead. Het verplaatsen van gegevens tussen clusters is echter nooit zo eenvoudig geweest.

Organisaties beheren meerdere Kafka-clusters om goede redenen: geografische spreiding, naleving van wet- en regelgeving (compliance), isolatie van teams of versie-scheiding. Maar zodra gegevens in een cluster landen, vereiste het maken van een getrouwe kopie in een ander cluster altijd externe tooling, zorgvuldige coördinatie en een grote tolerantie voor operationele verrassingen.

KIP-1279 verandert dit. Cluster mirroring integreert cross-cluster replicatie direct in de Kafka-broker. Geen externe processen, geen offset-vertaaltabellen en geen hercompressie. Een doelserver (destination broker) haalt committed records op uit een broncluster met hetzelfde fetch-protocol dat followers al gebruiken, en voegt deze byte voor byte toe aan de lokale partitielogs. Het resultaat is een mirror die offsets, compressie en de status van consumer groups behoudt, waardoor een failover zo simpel wordt als het stoppen van de mirror en het omleiden van clients.

Dit artikel behandelt de architectuur achter cluster mirroring, de state machine die de mirror-partities aanstuurt en de consistentiegaranties die het geheel ondersteunen. Daarna volgen twee praktische scenario's (disaster recovery en cluster-migratie), afgesloten met een video-demo.

Een nieuwe benadering van cross-cluster replicatie

Sinds Kafka 2.4 is MirrorMaker 2 (MM2) de standaardtool voor cross-cluster replicatie. MM2 draait als een set Kafka Connect-workers die gegevens consumeren van een broncluster en produceren naar een doelcluster. Cluster mirroring hanteert een fundamenteel andere aanpak door replicatie direct in de broker in te bedden.

• Geen externe infrastructuur: Mirror fetcher-threads draaien binnen het brokerproces zelf. Er hoeven geen Connect-workers te worden geprovisioneerd, gemonitord of geschaald. Een enkel CLI-commando (kafka-cluster-mirrors.sh --create) stelt een mirror in; een ander (--start) begint met het repliceren van topics. De volledige levenscyclus wordt beheerd via dezelfde Admin API die wordt gebruikt voor topics en consumer groups.
• Byte-voor-byte overdracht: Gecomprimeerde batches worden gerepliceerd als ruwe bytes, zonder ze te decompresseren of opnieuw te comprimeren. Een gzip, snappy, lz4 of zstd batch komt in zijn oorspronkelijke vorm aan bij de bestemming. Dit elimineert de CPU-overhead van een decompressie/recompressie-cyclus en behoudt de oorspronkelijke compressiekeuzes van de producer.
• Exacte offset-behoud: De doellog behoudt dezelfde offsets als de bron, inclusief gaten die zijn ontstaan door topic compaction. Consumer groups kunnen failover uitvoeren zonder offset-vertaling: de committed offset op de bron is gelijk aan de committed offset op de bestemming.
• Failover met één commando: Het stoppen van een mirror (--stop) brengt partities door een deterministische reeks: fetchers worden verwijderd, de laatste mirror epoch wordt opgeslagen, de leader epoch wordt verhoogd, openstaande transacties worden afgebroken en een nieuw control-record laat alle producer-status verlopen. De partitie wordt vervolgens schrijfbare op de bestemming. Externe coördinatie of offset-queries zijn niet nodig.
• Ondersteuning voor unclean leader election: Wanneer het broncluster een unclean leader election (ULE) ervaart, gaat de bestemming over in een herstelstatus. Deze wacht tot alle toegewezen replica's (niet alleen ISR-leden) convergeren naar de getrunceerde offset voordat de replicatie wordt hervat. Dit garandeert log-consistentie tussen clusters, zelfs als de bron een leader kiest met een onvolledige log.

De volgende tabel vat de belangrijkste verschillen samen:

Aspect	MirrorMaker 2	Cluster mirroring
Implementatie	Externe Connect-workers	Ingebouwd in de broker
Compressie	Decompresseren en hercompresseren	Byte-voor-byte doorvoer
Offsets	Verliesgevende vertaling via topic	Identiek over clusters
Consumer failover	Query offset sync topic	Direct, geen vertaling
Unclean elections	Geen afhandeling	Volledige log convergentie
Broncompatibiliteit	Kafka 2.0+	Kafka 2.1+
Monitoring	Connect-specifieke tooling	Standaard broker JMX-metrics

Bij cluster mirroring worden doelservers actieve deelnemers aan de cross-cluster replicatie. Elke server haalt gegevens direct op uit het broncluster via het standaard fetch-protocol en voegt ruwe record-batches toe aan lokale partitielogs. De bron- en doelpartities delen dezelfde topic ID.

Naast datareplicatie regelt de broker ook metadata-ontdekking, synchronisatie van configuraties, synchronisatie van group offsets en de propagatie van ACL's. Bandbreedtebeheer werkt aan beide kanten. De doelserver handhaaft een configureerbare limiet voor de replicatiesnelheid. Aan de bronzijde verschijnt het mirror-fetch-verkeer als standaard consumer-verzoeken, waardoor bestaande client-quota-mechanismen ongewijzigd van toepassing blijven.

Architectuur

Binnen elke doelserver werken drie hoofdcomponenten samen.

MirrorMetadataManager (MMM)

De MMM is de orchestrator. Deze draait op elke broker en implementeert de MetadataPublisher-interface om te reageren op wijzigingen in de KRaft-metadata log. Wanneer de controller een MirrorTopicStateChangeRecord schrijft, stuurt de MMM op de leader van de betreffende partitie de bijbehorende staatsovergang aan die een specifieke operatie triggert (create, start, stop, pause, resume, recover, delete).

De MMM onderhoudt ook een Admin-clientverbinding met het broncluster. Standaard ververst deze elke 60 seconden de bron-metadata: het ontdekken van nieuwe topics die voldoen aan de geconfigureerde include/exclude-patronen, het synchroniseren van topic-configuraties, het ophalen van consumer group offsets en het valideren dat de bron-cluster ID niet is gewijzigd. Die laatste controle voorkomt stille datacorruptie als iemand per ongeluk de mirror herconfigureert om naar een ander cluster te wijzen.

ClusterMirrorCoordinator (CMC)

De CMC regelt de persistentie van de status. Deze volgt hetzelfde coordinator-patroon als de group coordinator en de transaction coordinator, en beheert shards van een intern compacted topic genaamd __mirror_state (standaard: compact cleanup policy, 50 partities, replicatiefactor 3). De status van elke mirror-partitie wordt opgeslagen als een key-value record in dit topic, met optimistische concurrency-controle via leader epoch en state epoch fencing.

MirrorFetcherThread (MFT)

De MFT voert het zware werk uit. Door uitbreiding van Kafka's AbstractFetcherThread (dezelfde basisklasse die wordt gebruikt voor intra-cluster replicatie), haalt deze records op uit de bron en voegt deze toe aan lokale logs. Elke thread onderhoudt een dedicated NetworkClient met specifieke authenticatiegegevens per mirror, waardoor SASL/SSL-contexten tussen mirrors geïsoleerd blijven. De fetcher-manager indexeert threads op basis van een driedimensionale identifier (fetcher ID, bron-broker endpoint, mirror-naam), wat fijnmazige loadbalancing en een snelle reactie op leader-wijzigingen aan de bron mogelijk maakt.

Levenscyclus van een mirror-partitie

Een mirror-partitie doorloopt een reeks goed gedefinieerde statussen:

• LOG_ALIGNMENT: De eerste stap lijnt de lokale log uit met de bron. Als er geen LME (Last Mirror Epoch) wordt gevonden (bijv. bij eerste mirroring of een niet-ondersteunde bron), truncate de broker naar nul en start de mirroring vanaf nul. Anders wordt alleen het delta gerepliceerd. Het LME-record slaat de laatste mirror-offset en epoch van de vorige sessie op.
• EPOCH_FENCING: De broker stuurt een BumpLeaderEpochs-verzoek naar de controller, waarbij de lokale leader epoch met 10 wordt verhoogd. Zonder deze garantie zouden consumers op de bestemming kunnen initialiseren met een committed epoch van de bron die de lokale epoch overschrijdt, waardoor ze de leader als verouderd zouden weigeren.
• MIRRORING: De fetcher-thread begint records op te halen uit de bron. Gecomprimeerde batches worden direct aan de lokale log toegevoegd zonder recompressie. De high watermark schuift op terwijl lokale followers de data repliceren. Group offsets worden periodiek gesynchroniseerd vanaf de bron en beperkt tot het geldige offset-bereik van de bestemming.
• ULE_RECOVERY: Na de start van mirroring kan log-divergentie alleen worden veroorzaakt door een ULE in het broncluster, aangezien alleen committed records worden gemirrored. Wanneer ULE-ondersteuning is ingeschakeld en de bestemming-leader divergentie detecteert, gaat de partitie over naar deze status. De mirror-fetcher wordt verwijderd en het systeem wacht tot alle replica's convergeren naar de getrunceerde log-end offset.
• PAUSING / PAUSED: Een operator kan de mirroring pauzeren zonder voortgang te verliezen. Fetcher-threads worden beëindigd, maar de partitie blijft read-only. Het hervatten gaat direct terug naar MIRRORING.
• STOPPING / STOPPED: Dit is het failover-pad. De broker verwijdert fetcher-threads, registreert de LME voor toekomstige failback, verhoogt de lokale epoch, voegt ABORT-markers toe voor lopende transacties en schrijft een MIRRORPIDRESET control-record om alle producer-status te laten verlopen. De partitie wordt vervolgens schrijfbare.
• FAILED: Fouten triggeren een automatische retry met een exponentiële backoff. Niet-herstelbare fouten (bijv. wijziging van bron-cluster ID, verwijdering van bron-topic) vereisen handmatig herstel.

Dataconsistentie

Er zijn twee mechanismen die de log van het doelcluster consistent houden met de bron: log-convergentie en transactieveiligheid.

Log-convergentie

De destination leader epoch moet altijd groter zijn dan of gelijk zijn aan de source leader epoch (DLE $\ge$ SLE). Het systeem handhaaft dit gat via reactieve bumps (wanneer een fetched batch epoch de lokale epoch nadert), proactieve bumps (wanneer het gat onder de 3 zakt) en periodieke bumps (tijdens bron-metadata synchronisatie).

Wanneer mirroring start of herstart, kunnen de lokale en bron-logs zijn gedivergeerd. Een protocol in twee fasen lost dit op:

1. Initiële fase: Lijnt leader epoch-geschiedenissen uit via het LME-record.
2. Steady state fase: Lijnt offsets uit binnen resterende epochs tijdens normale fetch-verwerking. De fetcher detecteert dynamisch de fetch-versie van de bron. Bij Fetch v12+ maakt divergentie-informatie inline-truncatie mogelijk. Bij oudere versies wordt teruggevallen op een expliciet OffsetsForLeaderEpoch-verzoek.

Omdat de bestemming tijdens mirroring alleen committed data ophaalt, moet elke truncatie na LOGALIGNMENT voortkomen uit een ULE op de bron. De configuratie mirror.unclean.leader.election.enable bepaalt de reactie: bij true gaat de partitie naar ULERECOVERY, bij false (standaard) gaat de partitie naar FAILED.

Transactieveiligheid

Cluster mirroring repliceert met READ_UNCOMMITTED isolatie. Dit betekent dat onvoltooide transactie-records aankomen bij de bestemming voordat de bron over hun lot beslist. Dit is bewust gedaan om latentie en complexiteit in het datapad te voorkomen.

Tijdens de stopping-transitie voegt de broker ABORT-markers toe voor alle lopende transacties, zodat READ_COMMITTED consumers na de failover nooit "hangende" transacties zien. Ook wordt een MirrorPidResetRecord geschreven, zodat nieuwe producers op de bestemming verse producer-ID's kunnen verkrijgen zonder conflicten.

Disaster Recovery

Disaster recovery is de meest natuurlijke toepassing van cluster mirroring. Dit proces verloopt in drie fasen:

1. Normale werking

Cluster A bedient al het productieverkeer. Een mirror genaamd a-to-b repliceert continu topics naar Cluster B.

Op Cluster B wordt de mirror aangemaakt en gestart:

kafka-cluster-mirrors.sh --bootstrap-server B:9092 \
--create --mirror a-to-b --mirror-config mirror.properties

kafka-cluster-mirrors.sh --bootstrap-server B:9092 \
--start --mirror a-to-b --topics ".*"

De --describe flag kan worden gebruikt om de voortgang te monitoren:

kafka-cluster-mirrors.sh --bootstrap-server B:9092 \
--describe --mirror a-to-b

2. Failover

Cluster A valt uit. Eén commando op Cluster B stopt de mirroring en maakt alle topics schrijfbare:

kafka-cluster-mirrors.sh --bootstrap-server B:9092 \
--stop --mirror a-to-b

Producers en consumers schakelen hun bootstrap-servers over naar Cluster B. Omdat offsets identiek zijn en consumer group offsets zijn gesynchroniseerd, hervatten applicaties waar ze waren gebleven, zonder offset-vertaling of herverwerking.

3. Failback

Wanneer Cluster A herstelt, stelt de operator een reverse mirror in. Het systeem detecteert dat A voorheen de bron was, raadpleegt de LME uit de failover en voert een incrementele truncatie uit in plaats van de volledige dataset opnieuw te repliceren.

Op Cluster A wordt de reverse mirror gestart:

kafka-cluster-mirrors.sh --bootstrap-server A:9092 \
--create --mirror b-to-a --mirror-config reverse.properties

kafka-cluster-mirrors.sh --bootstrap-server A:9092 \
--start --mirror b-to-a --topics ".*"

Zodra de replicatie is bijgewerkt, vindt de definitieve overstap plaats:

kafka-cluster-mirrors.sh --bootstrap-server A:9092 \
--stop --mirror b-to-a

De Recovery Point Objective (RPO) hangt af van de replicatie-lag op het moment van uitval. Omdat cluster mirroring asynchroon is, gaan records die op A zijn geproduceerd maar nog niet naar B zijn gerepliceerd verloren bij een ongeplande failover.

Cluster-migratie

Cluster mirroring ondersteunt bron-brokers vanaf Kafka 2.1, wat een praktische migratiepad mogelijk maakt zonder tussenstappen van versie-upgrades.

De traditionele migratie vereist upgrades via elke grote versie (2.x → 3.x → 4.x) en een in-place ZooKeeper-naar-KRaft migratie. Met cluster mirroring is het proces als volgt:

1. Stel een nieuw KRaft-cluster op met de gewenste Kafka-versie.
2. Maak een mirror van het oude cluster naar het nieuwe.
3. Start de mirroring. De bestemming ontdekt automatisch topics, maakt matching partitions (met identieke topic ID's), synchroniseert configuraties en begint met repliceren.
4. Stop producers op het oude cluster.
5. Monitor de replicatie-lag tot deze nul is.
6. Stop de mirroring (topics op het nieuwe cluster worden schrijfbare).
7. Leid clients om naar het nieuwe cluster.
8. Decommission het oude cluster.

Commando's op het nieuwe cluster:

kafka-cluster-mirrors.sh --bootstrap-server new:9092 \
--create --mirror old-to-new --mirror-config old-cluster.properties

kafka-cluster-mirrors.sh --bootstrap-server new:9092 \
--start --mirror old-to-new --topics ".*"

Voortgang monitoren:

kafka-cluster-mirrors.sh --bootstrap-server new:9092 \
--describe --mirror old-to-new

Overstap bij nul lag:

kafka-cluster-mirrors.sh --bootstrap-server new:9092 \
--stop --mirror old-to-new

Conclusie

Cluster mirroring markeert een verschuiving in hoe Apache Kafka omgaat met cross-cluster databewegingen. Door replicatie in de broker in te bedden, elimineert KIP-1279 een hele klasse van operationele complexiteit en verbetert het de datatrouw: gecomprimeerde batches blijven intact, offsets blijven identiek en consumer failover vereist geen vertaling.

Voor disaster recovery vervangen twee commando's een uitgebreid runbook. Voor cluster-migraties biedt het een schoon upgrade-pad van ZooKeeper-deployments (vanaf Kafka 2.1) naar moderne KRaft-clusters. Toekomstige ontwikkelingen omvatten synchrone mirroring voor zero-RPO vereisten, integratie van tiered storage en ondersteuning voor diskless topics.