// Module included in the following assemblies:
//
// assembly-config.adoc

[id='con-config-mirrormaker2-connectors-{context}']
= Configuring MirrorMaker 2 connectors

[role="_abstract"]
MirrorMaker 2 uses internal connectors to synchronize data between Kafka clusters. 
Configure these connectors in your `KafkaMirrorMaker2` custom resource.

MirrorMaker 2 consists of the following connectors:

`MirrorSourceConnector`:: (Required) Replicates topics and ACLs from source to target cluster.
`MirrorCheckpointConnector`:: (Optional) Synchronizes consumer group offsets for failover support. Only needed if offset tracking is required.
`MirrorHeartbeatConnector`:: (Optional) Monitors connectivity between clusters. Useful for alerts but not required for replication. 

The following table describes connector properties and the connectors you configure to use them.

.MirrorMaker 2 connector configuration properties
[cols="4a,2,2,2",options="header"]
|===

|Property
|sourceConnector
|checkpointConnector
|heartbeatConnector


|admin.timeout.ms:: Timeout for admin tasks, such as detecting new topics. Default is `60000` (1 minute).
|✓
|✓
|✓

|replication.policy.class:: Policy to define the target topic naming convention. Default is `org.apache.kafka.connect.mirror.DefaultReplicationPolicy`. 
|✓
|✓
|✓

|replication.policy.separator:: The separator used for topic naming in the target cluster. By default, the separator is set to a dot (.). 
Separator configuration is only applicable to the `DefaultReplicationPolicy` replication policy class, which defines target topic names.
The `IdentityReplicationPolicy` class does not use the property as topics retain their original names.
|✓
|✓
|✓

|consumer.poll.timeout.ms:: Timeout when polling the source cluster. Default is `1000` (1 second).
|✓
|✓
|

|offset-syncs.topic.location:: The location of the `offset-syncs` topic, which can be the `source` (default) or `target` cluster.
|✓
|✓
|

|topic.filter.class:: Topic filter to select the topics to replicate. Default is `org.apache.kafka.connect.mirror.DefaultTopicFilter`.
|✓
|✓
|

|config.property.filter.class:: Topic filter to select the topic configuration properties to replicate. Default is `org.apache.kafka.connect.mirror.DefaultConfigPropertyFilter`.
|✓
|
|

|config.properties.exclude:: Topic configuration properties that should not be replicated. Supports comma-separated property names and regular expressions.
|✓
|
|

|offset.lag.max:: Maximum allowable (out-of-sync) offset lag before a remote partition is synchronized. Default is `100`.
|✓
|
|

|offset-syncs.topic.replication.factor:: Replication factor for the internal `offset-syncs` topic. Default is `3`.
|✓
|
|

|refresh.topics.enabled:: Enables check for new topics and partitions. Default is `true`.
|✓
|
|

|refresh.topics.interval.seconds:: Frequency of topic refresh. Default is `600` (10 minutes). By default, a check for new topics in the source cluster is made every 10 minutes.
You can change the frequency by adding `refresh.topics.interval.seconds` to the source connector configuration.
|✓
|
|

|replication.factor:: The replication factor for new topics. Default is `2`.
|✓
|
|

|sync.topic.acls.enabled:: Enables synchronization of ACLs from the source cluster. Default is `true`. For more information, see xref:con-config-mirrormaker2-{context}[].
|✓
|
|

|sync.topic.acls.interval.seconds:: Frequency of ACL synchronization. Default is `600` (10 minutes).
|✓
|
|

|sync.topic.configs.enabled:: Enables synchronization of topic configuration from the source cluster. Default is `true`.
|✓
|
|

|sync.topic.configs.interval.seconds:: Frequency of topic configuration synchronization. Default `600` (10 minutes).
|✓
|
|

|checkpoints.topic.replication.factor:: Replication factor for the internal `checkpoints` topic. Default is `3`.
|
|✓
|

|emit.checkpoints.enabled:: Enables synchronization of consumer offsets to the target cluster. Default is `true`.
|
|✓
|

|emit.checkpoints.interval.seconds:: Frequency of consumer offset synchronization. Default is `60` (1 minute).
|
|✓
|

|group.filter.class:: Group filter to select the consumer groups to replicate. Default is `org.apache.kafka.connect.mirror.DefaultGroupFilter`.
|
|✓
|

|refresh.groups.enabled:: Enables check for new consumer groups. Default is `true`.
|
|✓
|

|refresh.groups.interval.seconds:: Frequency of consumer group refresh. Default is `600` (10 minutes).
|
|✓
|

|sync.group.offsets.enabled:: Enables synchronization of consumer group offsets to the target cluster `__consumer_offsets` topic. Default is `false`.
|
|✓
|

|sync.group.offsets.interval.seconds:: Frequency of consumer group offset synchronization. Default is `60` (1 minute).
|
|✓
|

|emit.heartbeats.enabled:: Enables connectivity checks on the target cluster. Default is `true`.
|
|
|✓

|emit.heartbeats.interval.seconds:: Frequency of connectivity checks. Default is `1` (1 second).
|
|
|✓

|heartbeats.topic.replication.factor:: Replication factor for the internal `heartbeats` topic. Default is `3`.
|
|
|✓

|===

[WARNING] 
====
The following properties must be *identically configured across all three connectors* (source, checkpoint, and heartbeat):

* `replication.policy.class`
* `replication.policy.separator`
* `offset-syncs.topic.location`
* `topic.filter.class`

Mismatches cause replication failures or offset sync issues.
====