# Configure Cluster for Confluent Control Center Client Monitoring Set up your Apache Kafka® cluster for Control Center client metrics and to monitor share partition lag. ## Configure cluster for client metrics ### Prerequisites - Control Center running in normal mode - Minimum required Confluent Platform is 8.0 or higher - Minimum required client versions are Java client 3.7.2+ and librdkafka clients 2.6+ - Enable Telemetry Reporter on Confluent Platform ### Limitations - You must add a configuration to your self-managed connectors to enable client metrics for self-managed connectors. Add the following configuration to the connector configuration file: ```none producer.override.enable.metric.push=true ``` - Alternatively, you can use Control Center to add the configuration to enable client metrics for self-managed connectors. Add the configuration in **Additional Properties** of the connector for which you want to receive client metrics. For more information, see [Manage Kafka Connect](../connect.md#controlcenter-userguide-connect). - Without the `producer.override.enable.metric.push=true` configuration, Control Center does not display client metrics from self-managed connectors. - Some librdkafka clients do not expose all standard KIP-714 metrics. If you are working with such a client, some columns in Control Center are blank. This is due to this client limitation for some librdkafka clients. ### Configure cluster Use the following configurations to add and update the properties file of every Kafka broker in the cluster. Considerations: : - KRaft properties file: `server.properties` - File system location: `/kafka_2.13-3.8.0/config/kraft/server.properties` 1. Add the following configurations to the properties file of every Kafka broker. Telemetry Reporter configurations to add: ```none confluent.telemetry.external.client.metrics.push.enabled=true confluent.telemetry.external.client.metrics.delta.temporality=false confluent.telemetry.external.client.metrics.subscription.interval.ms.list=60000 confluent.telemetry.external.client.metrics.subscription.metrics.list=org.apache.kafka.consumer.fetch.manager.fetch.latency.avg,org.apache.kafka.consumer.connection.creation.total,org.apache.kafka.consumer.fetch.manager.fetch.total,org.apache.kafka.consumer.fetch.manager.bytes.consumed.rate,org.apache.kafka.producer.bufferpool.wait.ratio,org.apache.kafka.producer.record.queue.time.avg,org.apache.kafka.producer.request.latency.avg,org.apache.kafka.producer.produce.throttle.time.avg,org.apache.kafka.producer.connection.creation.total,org.apache.kafka.producer.request.total,org.apache.kafka.producer.topic.byte.rate ``` 2. Update the following configuration in the properties file of every Kafka broker. Telemetry Reporter configurations to update: ```none confluent.telemetry.exporter._c3.metrics.include=io.confluent.kafka.server.request.(?!.*delta).*|io.confluent.kafka.server.socket_server.connections|io.confluent.kafka.server.server.broker.state|io.confluent.kafka.server.replica.manager.leader.count|io.confluent.kafka.server.request.queue.size|io.confluent.kafka.server.broker.topic.failed.produce.requests.rate.1.min|io.confluent.kafka.server.tier.archiver.total.lag|io.confluent.kafka.server.request.total.time.ms.p99|io.confluent.kafka.server.broker.topic.failed.fetch.requests.rate.1.min|io.confluent.kafka.server.broker.topic.total.fetch.requests.rate.1.min|io.confluent.kafka.server.partition.caught.up.replicas.count|io.confluent.kafka.server.partition.observer.replicas.count|io.confluent.kafka.server.tier.tasks.num.partitions.in.error|io.confluent.kafka.server.broker.topic.bytes.out.rate.1.min|io.confluent.kafka.server.request.total.time.ms.p95|io.confluent.kafka.server.controller.active.controller.count|io.confluent.kafka.server.session.expire.listener.zookeeper.disconnects.total|io.confluent.kafka.server.request.total.time.ms.p999|io.confluent.kafka.server.controller.active.broker.count|io.confluent.kafka.server.request.handler.pool.request.handler.avg.idle.percent.rate.1.min|io.confluent.kafka.server.session.expire.listener.zookeeper.disconnects.rate.1.min|io.confluent.kafka.server.controller.unclean.leader.elections.rate.1.min|io.confluent.kafka.server.replica.manager.partition.count|io.confluent.kafka.server.controller.unclean.leader.elections.total|io.confluent.kafka.server.partition.replicas.count|io.confluent.kafka.server.broker.topic.total.produce.requests.rate.1.min|io.confluent.kafka.server.controller.offline.partitions.count|io.confluent.kafka.server.socket.server.network.processor.avg.idle.percent|io.confluent.kafka.server.partition.under.replicated|io.confluent.kafka.server.log.log.start.offset|io.confluent.kafka.server.log.tier.size|io.confluent.kafka.server.log.size|io.confluent.kafka.server.tier.fetcher.bytes.fetched.total|io.confluent.kafka.server.request.total.time.ms.p50|io.confluent.kafka.server.tenant.consumer.lag.offsets|io.confluent.kafka.server.session.expire.listener.zookeeper.expires.rate.1.min|io.confluent.kafka.server.log.log.end.offset|io.confluent.kafka.server.broker.topic.bytes.in.rate.1.min|io.confluent.kafka.server.partition.under.min.isr|io.confluent.kafka.server.partition.in.sync.replicas.count|io.confluent.telemetry.http.exporter.batches.dropped|io.confluent.telemetry.http.exporter.items.total|io.confluent.telemetry.http.exporter.items.succeeded|io.confluent.telemetry.http.exporter.send.time.total.millis|io.confluent.kafka.server.controller.leader.election.rate.(?!.*delta).*|io.confluent.telemetry.http.exporter.batches.failed|org.apache.kafka.consumer.(fetch.manager.fetch.latency.avg|connection.creation.total|fetch.manager.fetch.total|fetch.manager.bytes.consumed.rate)|org.apache.kafka.producer.(bufferpool.wait.ratio|record.queue.time.avg|request.latency.avg|produce.throttle.time.avg|connection.creation.total|request.total|topic.byte.rate) ``` 3. Restart every Kafka broker. ## Configure cluster to monitor share partition lag Share partition lag represents the number of records that have not yet been fully processed by consumers in a share group. It is the primary metric used to automate the scaling of share consumer instances. Unlike standard consumer lag, share partition lag accounts for the non-contiguous state of records in these states: * **AVAILABLE:** Not yet acquired by any consumer * **ACQUIRED:** Currently assigned to a consumer with an active acquisition lock * **ACKNOWLEDGED:** Processing complete * **ARCHIVED:** Removed by compaction Share partition lag is calculated as: ```text (highest offset) - (share-partition start offset) + 1 - (processed in-flight records) ``` To monitor lag effectively, you must distinguish between record states. Records in a **Terminal** state have been acknowledged or archived. These records no longer contribute to the lag calculation. To monitor share partition lag: 1. Set the following property in `KafkaBrokerConfig` to view share partition lag details: ```none confluent.share.lag.calculator.enabled=true ``` 2. Add these configurations to the `etc/kafka/broker.properties` file to override the default value of `3` for these configurations: ```none share.coordinator.state.topic.replication.factor=1 share.coordinator.state.topic.min.isr=1 ```