Using MirrorMaker for Replication

1. Understanding MirrorMaker Architecture

Component	Description	Role
Source Cluster	Original Kafka cluster from which data is replicated	Producer of messages to be mirrored
Target Cluster	Destination Kafka cluster where data is replicated	Consumer of mirrored messages
MirrorMaker	Application that reads from source and writes to target	Bridge between source and target clusters
Consumer Group	Group of MirrorMaker instances consuming from source	Ensures parallel processing and fault tolerance

Example: MirrorMaker Configuration

connectors:
  - name: "mirror-source"
    config:
      connector.class: "org.apache.kafka.connect.mirror.MirrorSourceConnector"
      tasks.max: 1
      source.cluster.alias: "source-cluster"
      target.cluster.alias: "target-cluster"

Note: MirrorMaker 2.0 uses Kafka Connect for replication, offering better scalability and reliability than MirrorMaker 1.0.

2. Configuring MirrorMaker 2.0

Example: Basic MirrorMaker 2.0 Configuration

{
  "name": "mirror-connector",
  "config": {
    "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "source.cluster.alias": "source-cluster",
    "target.cluster.alias": "target-cluster",
    "topics": "source-topic-1,source-topic-2",
    "replication.factor": 3
  }
}

Property	Description	Default Value	Notes
connector.class	Class name for the connector	None	Must be one of MirrorSourceConnector or MirrorHeartbeatConnector
source.cluster.alias	Alias for the source Kafka cluster	None	Used to identify the source cluster in logs and metrics
target.cluster.alias	Alias for the target Kafka cluster	None	Used to identify the target cluster in logs and metrics
topics	Topics to replicate	None	Comma-separated list of topic names or regex patterns
replication.factor	Replication factor for replicated topics	1	Must be at least 1 and less than or equal to the number of brokers
refresh.topics.interval.seconds	Interval to refresh topic list from source cluster	60	Used to detect new topics or topic deletions
sync.topics.enabled	Whether to sync topic configurations	true	Enables or disables synchronization of topic settings like retention policy

Note: MirrorMaker 2.0 uses Kafka Connect under the hood, so all standard Kafka Connect configuration options apply.

Warning: Ensure that the source and target clusters are properly configured to allow replication. Misconfiguration can lead to data loss or inconsistency.

3. Setting Source Cluster

Property	Value	Description
source.cluster.bootstrap.servers	`localhost:9092`	Bootstrap servers for the source Kafka cluster
source.cluster.security.protocol	`PLAINTEXT`	Security protocol used to communicate with the source cluster (e.g., PLAINTEXT, SSL)
source.cluster.sasl.mechanism	`PLAIN`	SASL mechanism for authentication (e.g., PLAIN, SCRAM-SHA-256)
source.cluster.sasl.username	`user1`	Username for SASL authentication
source.cluster.sasl.password	`password123`	Password for SASL authentication

Example: Source Cluster Configuration

source.cluster.bootstrap.servers=host1:9092,host2:9092
source.cluster.security.protocol=SSL
source.cluster.sasl.mechanism=SCRAM-SHA-256
source.cluster.sasl.username=admin
source.cluster.sasl.password=securePassword

4. Setting Target Cluster

Property	Values	Description
target.cluster.bootstrap.servers	`localhost:9092`	Bootstrap servers for the target Kafka cluster
target.cluster.security.protocol	`PLAINTEXT`	Security protocol for target cluster (PLAINTEXT, SSL, SASL_SSL)
target.cluster.sasl.mechanism	`PLAIN`	SASL mechanism for authentication (if security.protocol is SASL_SSL)
target.cluster.sasl.username	`user1`	Username for SASL authentication (if required)
target.cluster.sasl.password	`password123`	Password for SASL authentication (if required)

Example: Target Cluster Configuration

bootstrap.servers=target-kafka:9092
security.protocol=SSL
ssl.truststore.location=/path/to/truststore.jks
ssl.truststore.password=changeit
ssl.keystore.location=/path/to/keystore.jks
ssl.keystore.password=changeit
ssl.key.password=changeit

Note: Ensure that the target cluster has sufficient resources to handle replicated data and that network connectivity is stable between source and target clusters.

5. Configuring Topic Replication

Property	Values	Description
replication.factor	`3`	Number of replicas for each partition
min.insync.replicas	`2`	Minimum number of replicas that must acknowledge a write
unclean.leader.election.enable	`false`	Allow replicas that are not in sync to become leaders

Example: Topic Replication Configuration

{
  "replication.factor": 3,
  "min.insync.replicas": 2,
  "unclean.leader.election.enable": false
}

Note: Ensure min.insync.replicas is less than or equal to replication.factor for consistency.

Warning: Setting unclean.leader.election.enable to true can lead to data loss.

6. Setting Replication Flow

Property	Values	Description
replication.factor	`3`	Number of replicas for each topic partition
min.insync.replicas	`2`	Minimum number of replicas that must acknowledge a write
unclean.leader.election.enable	`false`	Allow replicas that are not in sync to become leaders

Example: Configuring Replication Flow

{
  "replication.factor": 3,
  "min.insync.replicas": 2,
  "unclean.leader.election.enable": false
}

Note: Ensure all brokers in the cluster have matching replication settings to maintain consistency.

7. Configuring Consumer Groups Sync

Property	Values	Description
consumer.group.id	`mirror-maker-group`	Unique identifier for the consumer group used by MirrorMaker
consumer.bootstrap.servers	`localhost:9092`	Bootstrap servers for the consumer to connect to
consumer.enable.auto.commit	`false`	Disables automatic offset commits to prevent duplicate consumption
consumer.auto.offset.reset	`earliest`	Sets offset to earliest if no committed offset exists
consumer.fetch.min.bytes	`1`	Minimum bytes to fetch in a single request
consumer.max.poll.records	`1000`	Maximum number of records to fetch per poll

Example: Consumer Group Configuration

consumer.group.id=mirror-maker-group
consumer.bootstrap.servers=localhost:9092
consumer.enable.auto.commit=false
consumer.auto.offset.reset=earliest
consumer.fetch.min.bytes=1
consumer.max.poll.records=1000

8. Setting ACLs Replication

Property	Description	Default Value	Notes
replication.factor	Replication factor for the topics created by MirrorMaker	1	Controls how many replicas of each topic are created
num.partitions	Number of partitions for the topics created by MirrorMaker	1	Defines how many partitions are used for the replicated topics
topic.rename.format	Format for renaming topics during replication	`{topic}`	Can be used to prefix or suffix replicated topics

Example: Configure ACLs for Replication

kafka-acls.sh --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:replicator \
  --operation Read \
  --topic '*' \
  --group '*' \
  --cluster

Example: Set Up Replication with ACLs

{
  "name": "mirror-maker-config",
  "config": {
    "topics.regex": ".*",
    "replication.factor": 3,
    "num.partitions": 12,
    "topic.rename.format": "replicated_{topic}"
  }
}

9. Configuring Heartbeat Connectors

Property	Default Value	Description
heartbeat.interval.ms	`30000`	Interval in milliseconds between heartbeat messages.
heartbeat.timeout.ms	`60000`	Timeout in milliseconds for heartbeat responses.
heartbeat.enabled	`true`	Flag to enable or disable heartbeat connectors.

Example: Heartbeat Configuration

{
  "heartbeat.interval.ms": 30000,
  "heartbeat.timeout.ms": 60000,
  "heartbeat.enabled": true
}

Note: Heartbeat connectors are essential for monitoring the health and status of Kafka MirrorMaker replication.

10. Monitoring Replication Lag

Property	Default Value	Description
replication.lag.max.ms	`30000`	Maximum allowed lag in milliseconds before a consumer is considered out of sync.
replication.lag.check.interval.ms	`10000`	Interval in milliseconds to check replication lag.
replication.lag.threshold	`5000`	Threshold in milliseconds to trigger an alert when lag exceeds this value.

Example: Configuring Lag Monitoring in MirrorMaker

{
  "replication.lag.max.ms": 30000,
  "replication.lag.check.interval.ms": 10000,
  "replication.lag.threshold": 5000
}

Note: The replication lag is measured as the time difference between the latest message in the source cluster and the latest message consumed by MirrorMaker in the target cluster.

Warning: If replication lag exceeds replication.lag.max.ms, MirrorMaker will stop consuming messages from the source cluster to prevent data loss.