KafkaIQ

10.2.4 – Replication Consistency Challenges

Module: M10 – Advanced Topics & Architecture
Section: 10.2 – Multi-Datacenter Replication & Geo Architecture
Goal: Understand and mitigate cross-cluster replication consistency issues

Introduction

Cross-cluster replication introduces consistency challenges due to network latency, partial failures, and asynchronous replication. This module covers the key challenges and mitigation strategies.

Challenge 1: Replication Lag

The Problem

Asynchronous replication = messages written to source cluster before replicated to target.

Scenario:

T0: Producer writes message to Cluster A (offset 1000)
T1: Consumer reads from Cluster A (sees offset 1000)
T2: MM2 replicates to Cluster B (still at offset 999)
T3: Consumer fails over to Cluster B
T4: Consumer reads from B (only sees offset 999, missing 1000)

Result: Data loss during failover (RPO = replication lag).

Mitigation Strategies

1. Monitor Replication Lag

kafka-consumer-groups.sh --describe \
  --group __mirror_maker \
  --bootstrap-server target:9092

Alert if lag > threshold (e.g., 1000 messages or 10 seconds).

2. Increase Replication Throughput

tasks.max = 8
producer.batch.size = 65536
producer.linger.ms = 10
consumer.fetch.min.bytes = 1048576

3. Accept RPO (acknowledge data loss window = lag time).

Challenge 2: Out-of-Order Messages

The Problem

Network variability causes messages to arrive out of order across clusters.

Scenario:

Cluster A:
  T0: Write message A (partition 0)
  T1: Write message B (partition 1)

Cluster B (after replication):
  T2: Receive message B first (network faster for partition 1)
  T3: Receive message A second

Result: Order reversed on Cluster B

Mitigation Strategies

1. Single Partition (if global ordering required)

partitions = 1  # Forces ordering, but limits throughput

2. Timestamp-Based Ordering

// Consumer sorts by timestamp before processing
messages.sort((a, b) -> Long.compare(a.timestamp(), b.timestamp()));

3. Accept Eventual Consistency (tolerate temporary disorder).

Challenge 3: Duplicate Messages

The Problem

MM2 failures can cause message duplication.

Scenario:

T0: MM2 replicates message (offset 1000) to Cluster B
T1: MM2 crashes before committing offset
T2: MM2 restarts, re-reads from last committed offset (999)
T3: Re-replicates message 1000 to Cluster B (duplicate)

Mitigation Strategies

1. Idempotent Processing (consumer handles duplicates)

Set<String> processedIds = new HashSet<>();

for (ConsumerRecord record : records) {
    if (processedIds.contains(record.key())) {
        continue;  // Skip duplicate
    }
    process(record);
    processedIds.add(record.key());
}

2. Exactly-Once MM2 Configuration (Kafka 2.6+)

exactly.once.support = enabled

3. Deduplication Topic (store processed message IDs).

Challenge 4: Tombstone Handling

The Problem

Tombstones (null-value messages) for log compaction may not replicate correctly.

Scenario:

Cluster A:
  T0: Write key="user-1", value="data"
  T1: Write key="user-1", value=null (tombstone, delete)

Cluster B (after replication):
  T2: Receive key="user-1", value="data"
  T3: Receive tombstone... or not? (depends on compaction timing)

Result: Deleted key still exists on Cluster B

Mitigation Strategies

1. Sync Topic Configs

sync.topic.configs.enabled = true
sync.topic.configs.interval.seconds = 60

2. Force Compaction Before Replication

kafka-log-dirs.sh --trigger-log-compaction \
  --topic user-profiles \
  --bootstrap-server cluster-a:9092

Challenge 5: Schema Evolution

The Problem

Schema changes on source cluster may break consumers on target cluster.

Scenario:

Cluster A:
  T0: Producer uses Schema V1 (fields: name, email)
  T1: Producer upgrades to Schema V2 (adds field: phone)

Cluster B Consumer:
  T2: Receives Schema V1 message (OK)
  T3: Receives Schema V2 message (fails if not backward compatible)

Mitigation Strategies

1. Schema Registry Replication

# Replicate Schema Registry schemas
schema.registry.urls = http://schema-registry-a:8081,http://schema-registry-b:8081

2. Backward/Forward Compatibility (enforce schema evolution rules).

3. Consumer Schema Negotiation (consumers handle multiple schema versions).

Challenge 6: Consumer Group Offset Sync

The Problem

Consumer offsets on source don't map directly to target (different topic names, offsets).

Scenario:

Cluster A:
  Consumer Group "payments" at offset 5000 on topic "orders"

Cluster B (after failover):
  Consumer Group "payments" needs offset on topic "cluster-a.orders"?

Mitigation Strategies

MM2 Checkpoint Connector (automatic offset translation):

connector.class = org.apache.kafka.connect.mirror.MirrorCheckpointConnector
emit.checkpoints.interval.seconds = 60

Checkpoint records store source-to-target offset mappings.

Challenge 7: Network Partitions

The Problem

Network split between clusters causes partial replication.

Scenario:

T0: Cluster A and B connected, replicating normally
T1: Network partition (A and B can't communicate)
T2: Producers write to both clusters independently
T3: Network heals
T4: Conflicting writes exist on both clusters

Mitigation Strategies

1. Monitor Network Health

# Heartbeat monitoring
kafka-console-consumer.sh \
  --topic heartbeats \
  --bootstrap-server target:9092

2. Circuit Breaker (stop writes if replication fails).

3. Conflict Resolution (see 10.2.3 for strategies).

Challenge 8: Time Skew

The Problem

Clock drift between clusters causes timestamp inconsistencies.

Scenario:

Cluster A clock: 2024-01-01 10:00:00
Cluster B clock: 2024-01-01 10:00:05 (5 seconds ahead)

Message from A replicated to B appears "in the past"

Mitigation Strategies

1. NTP Synchronization (keep clocks in sync).

2. Use Kafka Timestamps (not system time):

log.message.timestamp.type = LogAppendTime  # Broker timestamp, not producer

Best Practices

  1. Monitor Replication Lag: Alert if lag > threshold (messages or time).

  2. Idempotent Processing: Handle duplicate messages gracefully.

  3. Schema Compatibility: Enforce backward/forward compatible schemas.

  4. Offset Sync: Use MM2 checkpoint connector for consumer failover.

  5. NTP Sync: Keep cluster clocks synchronized (< 1 second skew).

  6. Network Monitoring: Detect and alert on network partitions.

  7. Accept Eventual Consistency: Design for asynchronous replication (no strong consistency).

Key Takeaways

  1. Replication Lag: Asynchronous replication = RPO > 0 (data loss window).

  2. Out-of-Order: Network variability causes message reordering.

  3. Duplicates: MM2 failures can duplicate messages (idempotent processing required).

  4. Tombstones: May not replicate correctly (sync compaction settings).

  5. Schema Evolution: Replicate schemas or enforce compatibility.

  6. Offset Sync: MM2 checkpoint connector translates offsets for failover.

  7. Network Partitions: Can cause conflicting writes (conflict resolution needed).

  8. Time Skew: Synchronize clocks with NTP.

What's Next?

Next Module: 10.2.5 – Geo-DR Patterns

References:

  • MirrorMaker 2 → 10.2.2
  • Active-active conflicts → 10.2.3