# Monitor Replicator for Confluent Platform The recommended way to monitor Replicator is with Confluent Control Center and the Replicator monitoring extension. For more information about the monitoring extension, see [Replicator monitoring extension](#replicator-monitoring-extension). Additionally, you can use the following tools to monitor Replicator: * Replicator metrics * Consumer metrics * Producer metrics * Replication lag ## Replicator monitoring extension The Confluent Replicator monitoring extension collects detailed metrics from Replicator tasks using an exposed REST API. These endpoints provide the following information: * Throughput - the number of messages replicated per second. * Message lag - the number of messages that have been produced to the origin cluster that have not yet been replicated to the destination. * Latency - the average time period between message production to the origin cluster and message production to the destination cluster. The metrics are broken down by connector, task, and topic/partition. Confluent designed these metrics to integrate with Control Center to provide a centralized view of replication within your clusters. The Replicator latency metric can’t report latency for Replicator when replicating historical data. It can only report latency when data is being replicated in real time. So when replicating very old data, you can disregard the latency reported by Replicator. ### Install The Monitoring extension is installed with Replicator. For more information about installation, see [Download and install Replicator](replicator-run.md#replicator-install). ### Activate the extension After the installation is complete, you must configure Kafka Connect to recognize the installed extension. First, add the extension library to the Connect worker classpath by setting the `CLASSPATH` environment variable. ```bash export CLASSPATH=/replicator-rest-extension-.jar ``` Then, to activate the extension, add the following property to the list of extension classes in the configuration file that you use to start Replicator (for example, `my-examples/replication.properties`, `etc/kafka/replication.properties`, or `/etc/kafka/connect-distributed.properties`): ```bash rest.extension.classes=io.confluent.connect.replicator.monitoring.ReplicatorMonitoringExtension ``` With the extension activated, you can use Control Center to monitor replicators. See [Use Control Center to monitor replicators](replicator-quickstart.md#rep-quickstart-monitoring) in the Replicator Tutorial and [Replicators](https://docs.confluent.io/control-center/current/replicators.html) in the Control Center User Guide. ### Additional configuration for secure endpoints When the Connect distributed cluster hosting Replicator exposes a REST endpoint with TLS/SSL authentication enabled (https), you must configure security properties for the TLS/SSL keystore and truststore used by the Replicator monitoring extension to communicate with other Connect nodes in the cluster. These are set with the following environment variables at the JVM level: ```bash export KAFKA_OPTS: -Djavax.net.ssl.trustStore=/etc/kafka/secrets/kafka.connect.truststore.jks -Djavax.net.ssl.trustStorePassword= -Djavax.net.ssl.keyStore=/etc/kafka/secrets/kafka.connect.keystore.jks -Djavax.net.ssl.keyStorePassword= ``` #### SEE ALSO For an example that shows how to set Docker environment variables for Confluent Platform running on KRaft, see the [Confluent Platform demo](../../tutorials/cp-demo/index.md#cp-demo). Refer to the demo’s [docker-compose.yml file](https://github.com/confluentinc/cp-demo/tree/latest/docker-compose.yml) for a configuration reference. You must also set `listeners.https.` properties, as specified in [TLS/SSL REST configuration](../../security/protect-data/encrypt-tls.md#encryption-ssl-rest), for example: ```bash "listeners.https.ssl.keystore.location": "/etc/kafka/connect.jks" "listeners.https.keystore.password": "xxxxx" ``` ### systemctl command configurations If you are not using `systemctl`, skip these steps. If you are using the `systemctl` command to start the monitoring service, you must configure the environment variables using `sudo systemctl edit confluent-kafka-connect`, or edit file `/etc/systemd/system/confluent-kafka-connect.service.d/`. If you use this method and don’t configure the environment variables properly for `systemctl`, Connect fails to start. The steps for this use case are as follows. 1. As previously described, activate the extension by adding the following property in the configuration file that you use to start Replicator (for example, `my-examples/replication.properties`, `etc/kafka/replication.properties`, or `/etc/kafka/connect-distributed.properties`): ```bash rest.extension.classes=io.confluent.connect.replicator.monitoring.ReplicatorMonitoringExtension ``` 2. Configure environment variables using either `sudo systemctl edit confluent-kafka-connect`, or by editing the file `/etc/systemd/system/confluent-kafka-connect.service.d/`: ```properties [Service] Environment="CLASSPATH=/replicator-rest-extension-.jar" ``` 3. Reload systemd configuration and restart the component. ```bash sudo systemctl reload sudo systemctl restart confluent-kafka-connect ``` ### Replicator monitoring extension API reference ### GET /WorkerMetrics/(string: connector_name) Get metrics for tasks running on this Connect worker for a given connector. * **Parameters:** * **connector_name** (*string*) – the name of the connector **Example request**: ```http GET /WorkerMetrics/replicate-topic HTTP/1.1 Host: kafkaconnect.example.com ``` **Example response**: ```http HTTP/1.1 200 OK Content-Type: application/json { "connectorName": "replicate-topic", "srcClusterId": "qhS9FOVFQkmokuIvo1ZZMA", "destClusterId": "qhS9FOVFQkmokuIvo1ZZMA", "tasks": [{ "id": "replicate-topic-0", "workerId": null, "state": null, "metrics": [{ "timestamp": 1570742038165, "throughput": 0.1075268817204301, "messageLag": 0.0, "latency": 12.764705882352942, "srcTopic": "wikipedia.parsed", "srcPartition": 1, "destTopic": "wikipedia.parsed.replica" }, { "timestamp": 1570742038166, "throughput": 0.10976948408342481, "messageLag": 0.0, "latency": 16.433333333333334, "srcTopic": "wikipedia.parsed", "srcPartition": 0, "destTopic": "wikipedia.parsed.replica" }] }] } ``` ### GET /ReplicatorMetrics Get metrics for all replicators running in this Connect cluster. **Example request**: ```http GET /ReplicatorMetrics HTTP/1.1 Host: kafkaconnect.example.com ``` **Example response**: ```http HTTP/1.1 200 OK Content-Type: application/json { "connectors": [{ "name": "replicate-topic", "srcClusterBootstrapServers": "kafka1:9091", "destClusterBootstrapServers": "kafka1:9091", "srcClusterId": "qhS9FOVFQkmokuIvo1ZZMA", "destClusterId": "qhS9FOVFQkmokuIvo1ZZMA", "tasks": [{ "id": "0", "workerId": "connect:8083", "state": "RUNNING", "metrics": [{ "timestamp": 1570742222631, "throughput": 0.11024142872891633, "messageLag": 0.0, "latency": 34.86666666666667, "srcTopic": "wikipedia.parsed", "srcPartition": 1, "destTopic": "wikipedia.parsed.replica" }, { "timestamp": 1570742222631, "throughput": 0.10187449062754686, "messageLag": 0.0, "latency": 15.307692307692308, "srcTopic": "wikipedia.parsed", "srcPartition": 0, "destTopic": "wikipedia.parsed.replica" }] }] }] } ``` ## Monitoring JMX metrics Like Apache Kafka® brokers, Kafka Connect reports metrics through JMX. To monitor Connect and Replicator, set the `JMX_PORT` environment variable before starting the Connect workers. Then, collect the reported metrics using your usual monitoring tools. JMXTrans, Graphite, and Grafana are a popular combination for collecting and reporting JMX metrics from Kafka. When you look at the metrics reported using JMX, Connect exposes Replicator’s consumer metrics and Connect’s producer metrics. You can view the full list of metrics in [Monitor Kafka with JMX in Confluent Platform](../../kafka/monitoring.md#kafka-monitoring). The following are some important metrics and their significance. ### Important Replicator metrics **MBean: confluent.replicator:type=confluent-replicator-task-metrics,confluent-replicator-task=([-.w]+),confluent-replicator-task-topic-partition=([-.w]+),confluent-replicator-name=([-.w]+),confluent-replicator-topic-name=([-.w]+)** `confluent-replicator-source-cluster` : The id of the source cluster from which Replicator is replicating. `confluent-replicator-destination-cluster` : The id of the destination cluster to which Replicator is replicating. `confluent-replicator-destination-topic-name` : The name of the destination topic to which Replicator is replicating. `confluent-replicator-task-topic-partition-message-lag` : The average number of messages that were produced to the origin cluster, but have not yet arrived to the destination cluster. If no new records are replicated for the partition for some time, this metric reports NaN. `confluent-replicator-task-topic-partition-throughput` : The number of messages replicated per second from the source to destination cluster. `confluent-replicator-task-topic-partition-byte-throughput` : The number of bytes replicated per second from the source to destination cluster. `confluent-replicator-task-topic-partition-latency` : The average time between message production to the source cluster and message production to the destination cluster. ### Important producer metrics **MBean: kafka.producer:type=producer-metrics,client-id=([-.w]+)** `io-ratio` or `io-wait-ratio` : If the `io-ratio` is low or `io-wait-ratio` is high, this means the producer is not very busy and is unlikely to be a bottleneck. `outgoing-byte-rate` : Reports the producer throughput when writing to destination Kafka. `batch-size-avg` and `batch-size-max` : If they are consistently close to the configured `batch.size`, you may be producing as fast as possible and you’ll want to increase the batch size to get better batching. `record-retry-rate` and `record-error-rate` : The average per-second number of retried record sends and failed record sends for a topic. A high number of those can indicate issues writing to the destination cluster. `produce-throttle-time-avg` and `produce-throttle-time-max` : Produce requests may be throttled to meet quotas configured on the destination cluster. If these are non-zero, it indicates that the destination brokers are slowing the producer down and you must review the quotas configuration. For more information about quotas, see [Enforcing client quotas](../../kafka/post-deployment.md#quotas). `waiting-threads` and `bufferpool-wait-time` : Non-zero values here indicate memory pressure. Connect producers can’t send messages fast enough, resulting in full memory buffers that cause Replicator threads to block. ### Important consumer metrics **MBean: kafka.consumer:type=consumer-fetch-manager-metrics,client-id=([-.w]+)** `io-ratio` or `io-wait-ratio` : If the `io-ratio` is low or `io-wait-ratio` is high, this means the consumer is not very busy and is unlikely to be a bottleneck. `bytes-consumed-rate` : Indicates the throughput of Replicator reading messages from the origin cluster. `fetch-size-avg` and `fetch-size-max` : If they are close to the configured maximum fetch size consistently, it means that Replicator is reading at the maximum possible rate. Increase the maximum fetch size and check if the throughput per task is improved. `records-lag-max` : The maximum lag in terms of number of records for any partition. An increasing value over time indicates that Replicator is not keeping up with the rate at which messages are written to the origin cluster. `fetch-rate`, `fetch-size-avg`, and `fetch-size-max` : If fetch-rate is high but fetch-size-avg and fetch-size-max are not close to the maximum configured fetch size, perhaps the consumer is “churning”. Try increasing the `fetch.min.bytes` and `fetch.max.wait` configuration. This can help the consumer batch more efficiently. `fetch-throttle-time-max` and `fetch-throttle-time-avg` : Fetch requests may be throttled to meet quotas configured on the origin cluster. If these are non-zero, it indicates that the origin brokers are slowing the consumer down and you must review the quotas configuration. For more information about quotas, see [Enforcing client quotas](../../kafka/post-deployment.md#quotas). ## Monitoring Replicator lag (legacy versions only) This section describes an alternate way to monitor Replicator lag using the consumer group lag tool, intended only for legacy versions of Replicator. #### IMPORTANT - For current versions of Replicator, it is recommended that you use the previously mentioned JMX metrics to monitor Replicator lag as it is more accurate than using the consumer group lag tool. If you’re using Replicator with a legacy version earlier than 5.4.0, this is the recommended methodology to monitor Replicator lag. - Replicator latency is calculated by taking the timestamp of the record consumed on the source and subtracting that from when the message offset to the destination is flushed. If old records are processed or if the time setting on the source records is not the same for producers and consumers, then the metric spikes. This spike is misleading and should not be construed as a latency issue, but rather is a limitation of this type of metrics calculation. You can monitor Replicator lag by using the [Consumer Group Command tool](https://docs.confluent.io/kafka/operations-tools/manage-consumer-groups.html) (`kafka-consumer-groups`). To use this functionality, you must set the Replicator `offset.topic.commit` config to `true` (the default value). Replicator doesn’t consume using a consumer group. Instead, it manually assigns partitions. When `offset.topic.commit` is true, Replicator commits consumer offsets (again manually), but these are for reference only and do not represent an active consumer group. Since Replicator only commits offsets and does not actually form a consumer group, the `kafka-consumer-groups` command output shows no active members in the group (correctly). It shows only the committed offsets. This is expected behavior for Replicator. To check membership information, use Connect status endpoints rather than `kafka-consumer-groups`. Replication lag is the number of messages that were produced to the origin cluster, but have not yet arrived to the destination cluster. It can also be measured as the amount of time it currently takes for a message to get replicated from origin to destination. If Replicator is behind for some reason and needs time to catch up, this can be higher than the latency between the two datacenters. The main reasons to monitor replication lag are: * If you need to fail over from origin to destination and the origin cannot be restored, all messages that were produced to the origin cluster and not replicated to the destination cluster are lost. (If the origin can be restored, the messages aren’t lost.) * The lag delays any event processing that happens at the destination. The lag is typically just a few hundred milliseconds (depending on the network latency between the two datacenters), but it can grow larger if network partitions or configuration changes temporarily pause replication and the replicator needs to catch up. If the replication lag keeps growing, it indicates that Replicator throughput is lower than what gets produced to the origin cluster and that additional Replicator tasks or Connect workers are necessary. For example, if producers are writing 100 MBps to the origin cluster but Replicator only replicates 50 MBps, the lag grows. To increase the throughput, the TCP socket buffer should be increased on the Replicator and the brokers. When Replicator is running in the destination cluster (recommended), you must also increase the following: - The TCP send socket buffer (`socket.send.buffer.bytes`) on the source cluster brokers. - The receive TCP socket buffer (`socket.receive.buffer.bytes`) on the consumers. A value of 512 KB is reasonable, but you may want to experiment with values up to 12 MB. If you are using Linux, you might need to change the default socket buffer maximum for the Kafka settings to take effect. For more information about tuning your buffers, see your operating system’s documentation for TCP tuning.