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

[id='con-mirrormaker-heartbeat-connector-{context}']
= Using the heartbeat connector to verify replication

[role="_abstract"]
You can enable the optional `MirrorHeartbeatConnector` to monitor the replication stream between clusters. 
When enabled, this connector periodically creates _heartbeat_ messages and sends them to a dedicated `heartbeats` topic in the _source_ cluster.

These messages are then replicated by the `MirrorSourceConnector` to the _target_ cluster. 
By consuming the `heartbeats` topic on the target cluster, you can confirm that data is flowing correctly from source to target. 
If you stop seeing new heartbeats on the target, it indicates a potential issue with the replication process.

Because the `MirrorHeartbeatConnector` writes heartbeat messages to the _source_ cluster, it usually requires a different Connect cluster from the one that runs the `MirrorSourceConnector` and `MirrorCheckpointConnector`. 

Before enabling the heartbeat connector in the `KafkaMirrorMaker2` resource, consider the advantages and disadvantages of running and maintaining an additional Kafka Connect cluster.
If you already operate a Kafka Connect cluster that stores state in the source Kafka cluster, you can deploy the `MirrorHeartbeatConnector` there instead of using the `KafkaMirrorMaker2` resource.

.Example configuration for the heartbeat connector defined in a `KafkaMirrorMaker2` resource
[source,yaml,subs="+attributes"]
----
apiVersion: {KafkaMirrorMaker2ApiVersion}
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
  # ...
  connectCluster: "my-cluster-target" # <1>
  clusters:
  - alias: "my-cluster-source"
    bootstrapServers: my-cluster-source-bootstrap:9092 # <2>
  - alias: "my-cluster-target"
    bootstrapServers: my-cluster-source-bootstrap:9092 # <3>
  mirrors:
    - sourceCluster: "my-cluster-source" # <4>
      targetCluster: "my-cluster-target" # <5>
      topicsPattern: "topic1|topic2|topic3"
      groupsPattern: "group1|group2|group3"
      # ...
      heartbeatConnector: # <6>
        autoRestart:
          enabled: true
        config:
          heartbeats.topic.replication.factor: 1 # <7>
          emit.heartbeats.interval.seconds: 10 # <8>
  # ...
----
<1> Alias of the Kafka Connect cluster where the connectors run. Must match the alias given in `targetCluster`.  
<2> Bootstrap address of the source cluster.  
<3> Bootstrap address of the target cluster. Use the same alias as the target cluster, but specify the bootstrap servers of the source cluster, because the heartbeat topic is created in the source.
<4> Alias of the source cluster. Must match the source alias defined for the `MirrorSourceConnector`.
<5> Alias of the target cluster. Must match the target alias defined for the `MirrorSourceConnector`.
<6> Configuration for the `MirrorHeartbeatConnector` that performs connectivity checks. Overrides the default configuration options.  
<7> Replication factor for the heartbeat topic.  
<8> Interval, in seconds, at which heartbeat messages are sent.  

IMPORTANT: The source and target cluster aliases in the `MirrorHeartbeatConnector` configuration must match the aliases defined for the `MirrorSourceConnector` in the `KafkaMirrorMaker2` resource.